Data Lake Formation:Manage multimodal autonomous driving data on DLF Paimon with Daft

Scenario

A self-driving fleet ingests multi-camera frame sequences into the lake to feed downstream training-set selection, regression sampling, and batch perception preprocessing. Common requirements:

• Locate frames quickly by channel (camera channel), time window, ego-vehicle state, and similar dimensions.

• Avoid materializing image BLOBs in full on every read.

• Tag training-set snapshots so you can reference them later by name.

Component responsibilities:

Prerequisites

CATALOG_OPTIONS = {
    "metastore": "rest",
    "uri": "http://<DLF-ENDPOINT>",       # Internet access must use HTTPS
    "warehouse": "<YOUR-CATALOG>",
    "token.provider": "dlf",
    "dlf.region": "<REGION-ID>",
    "dlf.access-key-id": "<ACCESS-KEY-ID>",
    "dlf.access-key-secret": "<ACCESS-KEY-SECRET>",
    "dlf.oss-endpoint": "<OSS-ENDPOINT>", # Optional over VPC; required over the Internet
}

Step 1: Prepare the source data

Load a public autonomous-driving road-scene dataset from ModelScope and extract the JPG bytes along with image dimensions.

import io
from PIL import Image
from modelscope.msdatasets import MsDataset

ds = MsDataset.load("modelscope/image_object_detection_auto_dataset", split="test")
print(f"loaded {len(ds)} road-scene JPGs from ModelScope public dataset")

def fetch_jpgs(n: int = 24):
    out = []
    for i in range(min(n, len(ds))):
        row = ds[i]
        with open(row["Input Image:FILE"], "rb") as f:
            jpg = f.read()
        with Image.open(io.BytesIO(jpg)) as im:
            w, h = im.size
        out.append({"title": row["Title"], "bytes": jpg, "w": w, "h": h})
    return out

raws = fetch_jpgs(24)

These are real-world JPGs, roughly 30-100 KB each.

Step 2: Create a partitioned BLOB table

channel is the typical filter dimension (for example, "give me a window of frames from one camera"), so making it the partition key lets partition pruning kick in automatically.

import pyarrow as pa
from pypaimon import CatalogFactory, Schema

catalog = CatalogFactory.create(CATALOG_OPTIONS)
catalog.create_database("default", True)

pa_schema = pa.schema([
    pa.field("sample_token",  pa.string(), nullable=False),     # frame unique ID
    pa.field("scene_token",   pa.string()),                      # shared by frames from one scene
    pa.field("channel",       pa.string()),                      # partition key
    pa.field("timestamp",     pa.timestamp("ms")),
    pa.field("ego_speed_kmh", pa.float64()),
    pa.field("weather",       pa.string()),
    pa.field("width",         pa.int32()),
    pa.field("height",        pa.int32()),
    pa.field("frame_jpg",     pa.large_binary()),                # source image (BLOB)
    pa.field("thumbnail",     pa.large_binary()),                # thumbnail (BLOB)
])
schema = Schema.from_pyarrow_schema(
    pa_schema,
    partition_keys=["channel"],
    options={
        "bucket": "-1",
        "file.format": "parquet",
        "row-tracking.enabled": "true",     # required for BLOB columns
        "data-evolution.enabled": "true",   # required for BLOB columns
    },
)
catalog.create_table("default.drive_frames", schema, True)

Step 3: Write the multimodal data

Synthesize nuScenes-style metadata (channel, scene_token, timestamp, ego_speed_kmh, weather) and fill the bytes field with the actual JPG bytes.

import daft
from datetime import datetime, timedelta
from pypaimon.daft import write_paimon

CHANNELS = ["CAM_FRONT", "CAM_FRONT_LEFT", "CAM_FRONT_RIGHT"]
WEATHERS = ["clear", "cloudy", "rain"]
base_ts  = datetime(2026, 5, 22, 9, 0, 0)

def make_thumbnail(jpg_bytes: bytes, max_side: int = 128) -> bytes:
    with Image.open(io.BytesIO(jpg_bytes)) as im:
        im.thumbnail((max_side, max_side))
        buf = io.BytesIO()
        im.convert("RGB").save(buf, format="JPEG", quality=70)
        return buf.getvalue()

rows = []
for i, r in enumerate(raws):
    rows.append({
        "sample_token":  r["title"],
        "scene_token":   f"scene_{i // 8:03d}",
        "channel":       CHANNELS[i % len(CHANNELS)],
        "timestamp":     base_ts + timedelta(milliseconds=i * 50),
        "ego_speed_kmh": float(20 + (i * 3) % 80),
        "weather":       WEATHERS[(i // 6) % len(WEATHERS)],
        "width":         r["w"],
        "height":        r["h"],
        "frame_jpg":     r["bytes"],
        "thumbnail":     make_thumbnail(r["bytes"]),
    })

arrow_tbl = pa.Table.from_pylist(rows, schema=pa_schema)
write_paimon(daft.from_arrow(arrow_tbl),
             "default.drive_frames", CATALOG_OPTIONS, mode="append")

Step 4: Explore the data across multiple dimensions

from pypaimon.daft import read_paimon

front = (
    read_paimon("default.drive_frames", CATALOG_OPTIONS)
    .where(daft.col("channel") == "CAM_FRONT")
    .sort("sample_token")
)
front.show()

fast = (
    read_paimon("default.drive_frames", CATALOG_OPTIONS)
    .where(daft.col("ego_speed_kmh") >= 60.0)
    .select("sample_token", "channel", "ego_speed_kmh", "weather")
    .sort("ego_speed_kmh", desc=True)
)
fast.show()

df = read_paimon("default.drive_frames", CATALOG_OPTIONS)
all_rows = df.to_pydict()
top3 = sorted(
    zip(all_rows["sample_token"], all_rows["channel"], all_rows["frame_jpg"]),
    key=lambda t: t[2].length,
    reverse=True,
)[:3]
for tk, ch, ref in top3:
    print(f"{tk} ({ch}): {ref.length} bytes, path={ref.path}")

d0c3fa90-d193ec71 (CAM_FRONT):       81852 bytes, path=oss://.../*.blob
d4316313-6a8d56d2 (CAM_FRONT_RIGHT): 81821 bytes, path=oss://.../*.blob
d4eb8adc-e95a5937 (CAM_FRONT_RIGHT): 80535 bytes, path=oss://.../*.blob

Step 5: Read BLOB bytes on demand

A daft.File is just a reference — it doesn't hold the bytes themselves. To consume the actual bytes, fetch them from OSS using path, offset, and length.

def fetch_blob(file_ref: daft.File, table) -> bytes:
    """Read the bytes referenced by a BLOB. `table` is the table object returned by catalog.get_table(...)."""
    with table.file_io.new_input_stream(file_ref.path) as stream:
        stream.seek(file_ref.offset)
        return stream.read(file_ref.length)

# Usage
table = catalog.get_table("default.drive_frames")
one = (
    read_paimon("default.drive_frames", CATALOG_OPTIONS)
    .where(daft.col("channel") == "CAM_FRONT")
    .limit(1).to_pydict()
)
ref = one["frame_jpg"][0]

raw = fetch_blob(ref, table)
print(f"fetched {len(raw)} bytes")

with Image.open(io.BytesIO(raw)) as im:
    print(f"decoded: {im.format} {im.size} {im.mode}")
# -> decoded: JPEG (1280, 720) RGB

Step 6: Tagging and time travel

Tag the current snapshot as training_v1 so you can return to this state later.

tbl = catalog.get_table("default.drive_frames")
snaps = sorted(tbl.snapshot_manager().list_snapshots(), key=lambda s: s.id)
print("snapshots:", [s.id for s in snaps])

tbl.create_tag("training_v1", snaps[0].id)

# Travel back to that point
df_v1 = read_paimon("default.drive_frames", CATALOG_OPTIONS, tag_name="training_v1")
print("training_v1 frames:", df_v1.count_rows())

You can also time travel by snapshot ID with read_paimon(..., snapshot_id=123).

Step 7: Plug in ML preprocessing as UDFs

enriched = (
    read_paimon("default.drive_frames", CATALOG_OPTIONS)
    .select("sample_token", "channel", "width", "height")
    .with_column("aspect_ratio", daft.col("width") / daft.col("height"))
)
enriched.show()

@daft.func(return_dtype=daft.DataType.string())
def speed_bucket(speed_col):
    out = []
    for v in speed_col.to_pylist():
        if v is None: out.append("unknown")
        elif v < 40:  out.append("low")
        elif v < 80:  out.append("mid")
        else:         out.append("high")
    return out

stats = (
    read_paimon("default.drive_frames", CATALOG_OPTIONS)
    .select("sample_token", "ego_speed_kmh")
    .with_column("speed_class", speed_bucket(daft.col("ego_speed_kmh")))
    .groupby("speed_class")
    .count("sample_token")
    .sort("speed_class")
)
stats.show()

high: 4 frames
low:  7 frames
mid:  13 frames

@daft.func(return_dtype=daft.DataType.float64())
def avg_brightness(file_col):
    out = []
    for ref in file_col.to_pylist():
        if ref is None:
            out.append(None); continue
        raw = fetch_blob(ref, table)              # goes through pypaimon FileIO
        with Image.open(io.BytesIO(raw)) as im:
            small = im.convert("L").resize((64, 64))
            px = list(small.getdata())
            out.append(sum(px) / len(px))
    return out

bri = (
    read_paimon("default.drive_frames", CATALOG_OPTIONS)
    .where(daft.col("channel") == "CAM_FRONT")
    .limit(4)
    .select("sample_token", "frame_jpg")
    .with_column("brightness", avg_brightness(daft.col("frame_jpg")))
    .select("sample_token", "brightness")
    .sort("brightness", desc=True)
)
bri.show()

cdbd1882-be82474a: 112.12
d0c3fa90-d193ec71: 111.34
cf0b73a9-3474b8ca:  34.71
cad180c4-553ceeb1:  33.93

Notes

• BLOB table schema constraint: a table that contains a large_binary column must set row-tracking.enabled=true and data-evolution.enabled=true, or the DLF service rejects the create-table request.

• Partition key choice: align the partition key with the typical filter (such as channel or capture_date) so that partition pruning actually fires.

• BLOB on-demand consumption: daft.File is a zero-copy reference and does not trigger an OSS read. The actual download only happens when you call table.file_io.new_input_stream(...).