Dask DataFrames

Install pipewhy[dask] to capture a trusted local Dask DataFrame. The first adapter accepts explicitly declared pandas int64, nullable Int64 and Python-backed string columns. The index is excluded. Other DataFrame types and remote cluster execution are not supported by this adapter.

from pathlib import Path
import dask
import dask.dataframe as dd
import pandas as pd
from pipewhy.adapters.dask import DaskSource
from pipewhy.cli.capture import scan_source
from pipewhy.formats import DaskLimits, DaskSchema, DaskWorkerLimits
from pipewhy.formats import PandasField, PandasLimits, PandasSchema, dask_source_version
from pipewhy.snapshots import LocalSnapshotStore

data = pd.DataFrame({"id": pd.array([1, 2, 3], dtype="Int64")})
with dask.config.set({"dataframe.convert-string": False}):
    frame = dd.from_pandas(data, npartitions=2, sort=False)
schema = DaskSchema(PandasSchema((PandasField(0, "id", "Int64"),)))
limits = DaskLimits(PandasLimits(1000, 10, 1024, 1000000, 100, 10000), 1000000, 10)
workers = DaskWorkerLimits(2147483648, 90000, 100000, 1000000)
source = DaskSource("people", frame, schema, limits, workers)
version = dask_source_version(frame, schema=schema, limits=limits, worker_limits=workers)
snapshot = scan_source(source, LocalSnapshotStore(Path("snapshots")), version)

Each partition executes in a fresh supervised local process with an explicit memory limit and deadline. The worker request is limited to 65,536 bytes. Graph bytes, partition count, records, fields, encoded staging, output frames and output bytes are bounded. The graph is prepared and serialized in the caller process before worker execution; its CPU time and transient serialization memory are not covered by the worker limit. The graph-byte limit bounds serialized output, not graph construction. During verification, retained records plus one additional partition can be in memory; the staging limit applies to each observation. Total runtime can include one deadline per partition per observation.

Only use trusted graphs: Dask tasks execute Python code, and the adapter stages serialized tasks temporarily for its workers. The temporary files are removed after each worker completes. Source versions include ordered partition results; capture repeats the observation to detect accidental changes before publishing. This does not guarantee an atomic read of mutable or nondeterministic sources. Strict normalization is the default; logical normalization is opt-in. Snapshots reopen without Dask or the original source.

Dask Arrays

One- and two-dimensional Dask Arrays use an explicit DaskArraySchema wrapping a NumpySchema. The first version supports signed and unsigned 64-bit integers, 64-bit floats and booleans. Chunk sizes must be known; a two-dimensional array must have a single chunk across its columns. Row chunks are executed in order, and each computed block must match its declared size. Structured arrays and masked arrays are rejected.

import dask.array as da
import numpy as np
from pipewhy.adapters.dask import DaskArraySource
from pipewhy.formats import DaskArrayLimits, DaskArraySchema, NumpyLimits, NumpySchema, dask_array_source_version

values = da.from_array(np.array([[1, 2], [3, 4]], dtype=">i8"), chunks=(1, 2))
array_schema = DaskArraySchema(NumpySchema((0, 1), "big", 2))
array_limits = DaskArrayLimits(NumpyLimits(1000, 10, 1000000, 100, 10000), 1000000, 10)
source = DaskArraySource("measurements", values, array_schema, array_limits, workers)
version = dask_array_source_version(values, schema=array_schema, limits=array_limits, worker_limits=workers)
snapshot = scan_source(source, LocalSnapshotStore(Path("snapshots")), version)

Array capture uses the same trusted-graph, worker, staging and repeated-read boundaries described above. Its schema identity is distinct from a DataFrame or a local NumPy array.