Skip to main content

Data Hashing and Cache Key Computation

Choose the cache-key strategy​

Use a whole-object HashFunction for a dataframe, an incremental HashlibAccumulator for file bytes, or a trusted string/PrecomputedValue when the cache key is already known. The public custom-hashing entry point is:

from flyte.io import HashFunction

All of these methods follow the HashMethod interface from flyte.io._hashing_io: update(data) accepts the value to hash and result() returns a string. The important distinction is what update means:

  • HashFunction calls the supplied function for each update and replaces its stored value. It is appropriate when the complete dataframe is passed in once.
  • HashlibAccumulator forwards each update to a stateful hashlib object. It is appropriate when a file is transferred in chunks.
  • PrecomputedValue ignores updates and always returns the string supplied to its constructor.
  • Passing a string as hash_method is also treated as a precomputed value by the dataframe and file APIs.

Create a new stateful accumulator for each independent file or dataframe. HashlibAccumulator retains its underlying hashlib state, and HashFunction.result() raises AttributeError until update() has been called.

Hash dataframes with a custom function​

Compute the hash in DataFrame.from_local​

DataFrame.from_local accepts hash_method: HashMethod | str | None. With a HashMethod, it calls update(df) once and stores result() as the dataframe hash; with a string, it stores the string directly.

import pandas as pd
from flyte.io._dataframe.dataframe import DataFrame
from flyte.io import HashFunction


def hash_pandas_dataframe(df: pd.DataFrame) -> str:
    return str(pd.util.hash_pandas_object(df).sum())


@env.task
async def produce_dataframe() -> DataFrame:
    df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
    hash_method = HashFunction.from_fn(hash_pandas_dataframe)
    return await DataFrame.from_local(df, hash_method=hash_method)

HashFunction.from_fn is a convenience factory around HashFunction(fn). The callable receives the dataframe itself and must return a string. The pandas example in examples/advanced/hash_flyte_dataframe.py uses pd.util.hash_pandas_object(df).sum() for a content-based value.

Do not use a HashFunction as though it were a chunk accumulator: every call to update invokes the function and overwrites the previous _value. For a dataframe, the callable should compute the hash of the complete object passed to it.

Attach a hash function to a dataframe annotation​

For a reusable task type, attach one HashFunction instance to typing.Annotated:

from typing import Annotated

import pandas as pd
from flyte.io import HashFunction


def hash_pandas_dataframe(df: pd.DataFrame) -> str:
    return str(pd.util.hash_pandas_object(df).sum())


HashedPandasDataFrame = Annotated[pd.DataFrame, HashFunction.from_fn(hash_pandas_dataframe)]


@env.task
async def produce_dataframe() -> HashedPandasDataFrame:
    df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
    return df

extract_cols_and_format in src/flyte/io/_dataframe/dataframe.py scans Annotated metadata and recognizes an object with isinstance(annotation, HashMethod). The dataframe transformer later applies that method during raw dataframe literal conversion. Only one hash method may appear in the annotation; multiple hash methods raise ValueError.

The same pattern is used for Polars. The repository example defines its callable with DataFrame.hash_rows():

from typing import Annotated

import polars as pl
from flyte.io import HashFunction


def hash_polars_dataframe(df: pl.DataFrame) -> str:
    return str(df.hash_rows().sum())


HashedPolarsDataFrame = Annotated[pl.DataFrame, HashFunction.from_fn(hash_polars_dataframe)]


@env.task
async def produce_dataframe() -> HashedPolarsDataFrame:
    df = pl.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
    return df

The dataframe examples add flyte.ctx().action.run_name to their data when they want hashes to differ between runs. If stable content-based cache keys are required, the callable must receive deterministic dataframe content instead.

Hash file uploads incrementally​

Use HashlibAccumulator.from_hash_name when the cache key should be computed from the bytes transferred:

from flyte.io import File
from flyte.io._hashing_io import HashlibAccumulator


@env.task
def upload_file() -> File:
    accumulator = HashlibAccumulator.from_hash_name("sha256")
    return File.from_local_sync(
        "/tmp/data.csv",
        "s3://my-bucket/data.csv",
        hash_method=accumulator,
    )

HashlibAccumulator.from_hash_name("sha256") calls hashlib.new("sha256"). Its update(memoryview) forwards each byte range to the hashlib object, and result() returns hexdigest().

The synchronous File.from_local_sync path uses HashingWriter while shutil.copyfileobj copies the source. For a remote destination, the writer wraps the filesystem destination; for a copied local destination, it wraps the local destination. The resulting File.hash is the accumulator result.

The asynchronous remote path uses AsyncHashingReader around the local source and passes that reader to storage.put_stream:

from flyte.io import File
from flyte.io._hashing_io import HashlibAccumulator


@env.task
async def upload_file() -> File:
    accumulator = HashlibAccumulator.from_hash_name("sha256")
    return await File.from_local(
        "/tmp/data.csv",
        "s3://my-bucket/data.csv",
        hash_method=accumulator,
    )

File.from_local and File.from_local_sync require the local path to exist and raise ValueError otherwise. A local-storage shortcut can avoid copying, so inspect the returned File.hash when using a local destination and expecting computed content hashing.

Supply an already-known file key​

Use a string when the API should assign a known key without computation:

from flyte.io import File


@env.task
async def reference_file() -> File:
    return await File.from_local(
        "/tmp/data.csv",
        "s3://my-bucket/data.csv",
        hash_method="verified-content-key",
    )

Use PrecomputedValue when a HashMethod object is required by surrounding code:

from flyte.io import File
from flyte.io._hashing_io import PrecomputedValue


@env.task
def upload_with_known_key() -> File:
    key = PrecomputedValue("my-precomputed-hash")
    return File.from_local_sync(
        "/tmp/data.bin",
        "s3://my-bucket/data.bin",
        hash_method=key,
    )

PrecomputedValue.update() is a no-op. File upload code detects it and transfers the file without a hashing reader or writer, then assigns result() to File.hash. It does not verify that the transferred bytes match the supplied key, so use it only for a key established elsewhere.

File.new_remote accepts the same two forms. A string becomes the known File.hash; a HashMethod is retained and is used while the returned remote file is written.

Set a directory cache key​

Dir currently takes a cache key as a string rather than accepting a HashMethod or hashing directory contents through the reader/writer utilities:

from flyte.io import Dir


@env.task
async def use_directory() -> Dir:
    return await Dir.from_local(
        "/tmp/data-dir/",
        dir_cache_key="directory-content-key",
    )

For an existing remote directory, assign the key with dir_cache_key as well:

from flyte.io import Dir


def existing_directory() -> Dir:
    return Dir.from_existing_remote(
        "s3://my-bucket/data/",
        dir_cache_key="directory-content-key",
    )

Both methods assign the supplied string directly to Dir.hash. The directory tests verify that from_existing_remote preserves the remote path and assigns dir_cache_key, and that from_local assigns its dir_cache_key to the returned directory. Dir.from_local also supports an optional batch_size; when omitted, its docstring specifies the FLYTE_IO_BATCH_SIZE setting, whose default is 32.

Use the hashing stream wrappers directly​

The wrappers in flyte.io._hashing_io are private implementation infrastructure. They all accept an underlying handle and a HashMethod, expose result(), and delegate public handle attributes. They are useful when implementing a transfer path that must hash exactly what it reads or writes.

Write and hash synchronously​

HashingWriter.write converts the supplied value to a memoryview, updates the accumulator, and then writes the original value to the wrapped handle. Hashing therefore happens before the underlying write:

import io

from flyte.io._hashing_io import HashingWriter, HashlibAccumulator


handle = io.BytesIO()
accumulator = HashlibAccumulator.from_hash_name("sha256")
writer = HashingWriter(handle, accumulator)
writer.write(b"hello")
writer.write(b" world")
assert writer.result() == accumulator.result()

Bytes, bytearray, memoryview, and other buffer-protocol objects are accepted. For str, the wrapper uses the explicit encoding, then the wrapped handle's encoding, then UTF-8 for hashing only; the original string is still passed to the handle. writelines calls write for every line, while flush, close, and public attributes delegate to the wrapped handle.

Write and hash asynchronously​

AsyncHashingWriter has the same conversion and update order, but write and writelines are awaitable:

from flyte.io._hashing_io import AsyncHashingWriter, HashlibAccumulator


async def write_data(async_handle) -> str:
    accumulator = HashlibAccumulator.from_hash_name("sha256")
    writer = AsyncHashingWriter(async_handle, accumulator)
    await writer.write(b"hello")
    await writer.writelines([b" ", b"world"])
    await writer.flush()
    return writer.result()

Its flush and close methods accept wrapped handles whose corresponding methods are either synchronous or awaitable. The current File.from_local implementation does not directly use AsyncHashingWriter; its asynchronous remote-upload path uses AsyncHashingReader instead.

Read and hash synchronously​

HashingReader updates the accumulator for data returned by read, readline, readlines, and iteration. It returns the original data to the caller:

import io

from flyte.io._hashing_io import HashingReader, HashlibAccumulator


reader = HashingReader(
    io.BytesIO(b"first\nsecond\n"),
    HashlibAccumulator.from_hash_name("sha256"),
)
for line in reader:
    print(line)
content_hash = reader.result()

Empty and None results are not sent to the accumulator. Text returned by a text-mode handle is encoded for hashing only. The reader hashes only bytes actually returned through its intercepted operations: mixing partial reads and iteration, or calling a delegated method that reads internally, can leave a partial-content hash.

Read and hash asynchronously​

AsyncHashingReader provides the corresponding async operations and is the wrapper used by asynchronous remote file uploads:

from flyte.io._hashing_io import AsyncHashingReader, HashlibAccumulator


async def read_data(async_handle) -> str:
    reader = AsyncHashingReader(
        async_handle,
        HashlibAccumulator.from_hash_name("sha256"),
    )
    while True:
        chunk = await reader.read(1024)
        if not chunk:
            break
    return reader.result()

read, readline, and readlines are awaited on the underlying handle. If the handle has no readlines, the wrapper falls back to async iteration. Async iteration prefers the underlying __anext__; otherwise it repeatedly calls readline and treats "" and b"" as end-of-stream. The primary read methods must be awaitable, while flush and close support either sync or async underlying methods.

Troubleshoot cache-key mismatches​

  • The same file has different keys: check the bytes actually supplied to write or returned by read, and check text encoding. The selection order is explicit encoding, handle encoding, then UTF-8; errors also affects text encoding.
  • A digest contains multiple objects: create a fresh HashlibAccumulator for each file. Reusing one instance continues the same hashlib state.
  • The digest is incomplete: consume the entire reader stream. Reader wrappers hash only intercepted data that was actually returned.
  • A custom dataframe hash is unexpectedly replaced: HashFunction.update is replacement-style, not incremental. Hash the complete dataframe in one callable invocation.
  • An annotation fails: extract_cols_and_format checks metadata for HashMethod instances and raises ValueError if more than one hash method is present.
  • A precomputed key does not match content: neither a string hash_method nor PrecomputedValue validates transferred content.
  • Caching does not appear during local runs: the advanced examples note that caching requires a remote Flyte cluster. File and dataframe upload behavior also depends on Flyte initialization, raw-data configuration, and storage access. Pandas examples require pandas and PyArrow in the task image; the Polars example requires the Flyte Polars plugin shown in its task-image configuration.