Data Lake Formation：使用 PyPaimon 讀取多模態表

少量資料一次性讀取

當您需要讀取單個 clip 或少量 clip，且結果能夠放入記憶體時，可以使用 read_blobs 方法一次性將資料物化到記憶體。

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,
    )
)

大批量流式讀取

當需要讀取大批量多 clip 資料時，推薦使用 stream_blobs 方法分 batch 流式消費，避免將大量 Blob bytes 一次性載入到記憶體。

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)

參數說明

訓練讀取連續 frame range

訓練取數情境下，通常需要從同一個 clip 中讀取連續的 frame window（如 16、32 或 64 幀）。以下樣本展示如何讀取單個 clip 內的一段連續 frame range。

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)

基本用法

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，包含 clip_id、frame_index 等標量列。
    blobs: dict[str, list[bytes | None]]，key 為 blob_cols 中的列名。
    傳回值需要是一個小的 pyarrow.Table；只做 side-effect 時可返回空表。
    """
    camera_0 = blobs["camera_0"]
    # 不建議返回原始 Blob bytes，避免把大量資料物化到 Ray 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,
)

# 觸發執行。Ray Dataset 是 lazy 的，不消費 result_ds 就不會真正讀取 Blob。
for _ in result_ds.iter_batches(batch_format="pyarrow"):
    pass

參數說明

配置 Ray task 重試

大規模冷讀情境下，建議為 map_with_blobs() 配置 Ray task 重試。

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,
    },
)

如果 descriptor 規模也較大，也可以為 to_ray() 傳入相同的 ray_remote_args。

基本用法

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 會把 Daft File/Blob descriptor 列讀取成 binary bytes。
# max_concurrency 是每個 Daft batch UDF 內部讀取 Blob 的並發度。
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)
    ],
)

# 樣本：這裡只統計 bytes 大小。真實業務中可接 decode、preprocess、inference、write 等邏輯。
# 不建議長期保留或寫出原始 Blob bytes，避免大量資料進入 Ray 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_blob 逐個開啟 Blob 流

如果需要在自訂 Daft UDF 中逐個開啟 Blob 流進行處理，可以使用 open_blob() 方法。其中 max_concurrency 控制每個 Daft batch UDF 內部讀取 Blob 的並發度。

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)