Data Lake Formation:Read DLF multimodal tables using PyPaimon

Read a small dataset at once

To read one clip or a few clips that fit in memory, use read_blobs to load all data at once.

scalar, blobs = (
    table.scan()
    .where("clip_id = 'xxx'")
    .select(["clip_id", "frame_index", "camera_0", "camera_1", "camera_2", "camera_3"])
    .read_blobs(
        ["camera_0", "camera_1", "camera_2", "camera_3"],
        parallelism=16,
    )
)

Stream large volumes of data

To read many clips without loading all BLOB bytes into memory at once, use stream_blobs to consume data in batches.

clip_ids = ["clip_001", "clip_002", "clip_003"]
clip_filter = "clip_id IN (" + ", ".join(f"'{clip_id}'" for clip_id in clip_ids) + ")"

for scalar_batch, blobs in (
    table.scan()
    .where(clip_filter)
    .select(["clip_id", "frame_index", "camera_0", "camera_1", "camera_2", "camera_3"])
    .stream_blobs(
        ["camera_0", "camera_1", "camera_2", "camera_3"],
        parallelism=16,
    )
):
    consume_batch(scalar_batch, blobs)

Parameters

Read consecutive frame ranges for training

Training workloads often require a consecutive frame window — for example, 16, 32, or 64 frames — from the same clip. The following example reads a consecutive frame range within a single clip.

start = 1000
read_frames = 512

for scalar_batch, blobs in (
    table.scan()
    .where(
        f"clip_id = 'xxx' "
        f"AND frame_index >= {start} "
        f"AND frame_index < {start + read_frames}"
    )
    .select(["clip_id", "frame_index", "camera_0", "camera_1", "camera_2", "camera_3"])
    .stream_blobs(
        ["camera_0", "camera_1", "camera_2", "camera_3"],
        parallelism=16,
    )
):
    consume_training_range(scalar_batch, blobs)

Basic usage

import ray
import pyarrow as pa

ray.init(address="auto", ignore_reinit_error=True)

clip_ids = ["clip_001", "clip_002", "clip_003"]
scalar_cols = ["clip_id", "frame_index"]
blob_cols = ["camera_0", "camera_1", "camera_2", "camera_3"]
select_cols = scalar_cols + blob_cols
clip_filter = "clip_id IN (" + ", ".join(f"'{clip_id}'" for clip_id in clip_ids) + ")"

def process_batch(scalar_batch, blobs):
    """
    scalar_batch: pyarrow.Table containing scalar columns such as clip_id and frame_index.
    blobs: dict[str, list[bytes | None]], keyed by blob column name.
    Must return a small pyarrow.Table. Return an empty table for side-effect-only processing.
    """
    camera_0 = blobs["camera_0"]
    # Avoid returning raw BLOB bytes to prevent materializing large payloads in Ray's object store.
    return pa.table({
        "rows": [scalar_batch.num_rows],
    })

ds = (
    table.scan()
    .where(clip_filter)
    .select(select_cols)
    .to_ray()
)

result_ds = table.map_with_blobs(
    ds,
    blob_cols,
    process_batch,
)

# Trigger execution. Ray Dataset is lazy — BLOBs are not read until result_ds is consumed.
for _ in result_ds.iter_batches(batch_format="pyarrow"):
    pass

Parameters

Configure Ray task retries

For large-scale cold-read scenarios, configure Ray task retries for map_with_blobs().

result_ds = table.map_with_blobs(
    ds,
    blob_cols,
    process_batch,
    parallelism=8,
    batch_size=512,
    ray_remote_args={
        "max_retries": 3,
        "retry_exceptions": True,
    },
)

If the descriptor volume is also large, pass the same ray_remote_args to to_ray().

Basic usage

import datetime as dt

import daft
from daft import col, runners
import pyarrow as pa
import ray

from pypaimon.daft import read_paimon, read_blob

ray.init(address="auto", ignore_reinit_error=True)
runners.set_runner_ray(address="auto", noop_if_initialized=True)

table_identifier = "<DATABASE_NAME>.<TABLE_NAME>"

clip_ids = ["clip_001", "clip_002", "clip_003"]
scalar_cols = ["clip_id", "frame_index", "collected_date"]
blob_cols = ["camera_0", "camera_1", "camera_2", "camera_3"]
target_date = dt.date(2026, 7, 1)

df = read_paimon(table_identifier, catalog_options)

df = df.where(
    (col("collected_date") == target_date)
    & col("clip_id").is_in(clip_ids)
)

# read_blob reads Daft File/Blob descriptor columns into binary bytes.
# max_concurrency controls read concurrency within each Daft batch UDF.
blob_bytes_cols = []
for blob_col in blob_cols:
    bytes_col = blob_col.replace(".", "_") + "_bytes"
    blob_bytes_cols.append(bytes_col)

df = df.select(
    *[col(name) for name in scalar_cols],
    *[
        read_blob(
            col(blob_col),
            catalog_options,
            table_identifier,
            max_concurrency=16,
        ).alias(bytes_col)
        for blob_col, bytes_col in zip(blob_cols, blob_bytes_cols)
    ],
)

# Example: aggregate byte sizes only. In production, chain decode, preprocess, inference, or write logic.
# Avoid retaining or writing raw BLOB bytes to prevent large payloads from entering Ray's object store.
agg_exprs = [col("clip_id").count().alias("rows")]
for bytes_col in blob_bytes_cols:
    agg_exprs.append(col(bytes_col).count().alias(bytes_col + "_count"))
    agg_exprs.append(col(bytes_col).str.length().sum().alias(bytes_col + "_sum"))

result = df.agg(*agg_exprs).collect()

Open individual blob streams with open_blob

To open and process BLOBs individually inside a custom Daft UDF, use open_blob(). The max_concurrency parameter controls read concurrency within each Daft batch UDF.

from pypaimon.daft import open_blob

def consume_one_blob(file):
    with open_blob(file, catalog_options, table_identifier) as stream:
        data = stream.read()
        # decode / preprocess / inference / write
        return len(data)