Data Lake Formation：基於Daft和DLF的自動駕駛多模態資料湖實踐

情境

智能駕駛車隊從多網路攝影機採集幀序列入湖，下遊用於訓練集篩選、迴歸集抽樣、感知模型批量預先處理。典型需求如下：

• 按 channel（相機通道）、時間窗、自車狀態等維度快速定位幀。

• BLOB 形式的映像不能每次都全量物化。

• 訓練集快照（snapshot）可回溯、可打 tag（tag 即 snapshot 的命名引用）。

各組件分工如下：

前提條件

CATALOG_OPTIONS = {
    "metastore": "rest",
    "uri": "http://<DLF-ENDPOINT>",       # 公網訪問須使用 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>", # VPC 可選，公網必填
}

步驟 1：準備素材資料

從 ModelScope 載入公開的自動駕駛道路情境資料集，並提取 JPG 原圖及其尺寸資訊。

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)

每張圖片約為 30-100KB JPG，均為真實公開資料。

步驟 2：建立 BLOB 分區表

channel 是典型查詢過濾維度（如“取某個相機的某段幀”），將其用作 partition key 可讓分區裁剪天然生效。

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),     # 幀唯一 ID
    pa.field("scene_token",   pa.string()),                      # 同一段情境的幀共用
    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()),                # 原圖（BLOB）
    pa.field("thumbnail",     pa.large_binary()),                # 縮圖（BLOB）
])
schema = Schema.from_pyarrow_schema(
    pa_schema,
    partition_keys=["channel"],
    options={
        "bucket": "-1",
        "file.format": "parquet",
        "row-tracking.enabled": "true",     # BLOB 列必需
        "data-evolution.enabled": "true",   # BLOB 列必需
    },
)
catalog.create_table("default.drive_frames", schema, True)

步驟 3：寫入多模態資料

合成 nuScenes 風格的中繼資料（channel、scene_token、timestamp、ego_speed_kmh、weather），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")

步驟 4：按維度探索資料

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

步驟 5：按需讀取 BLOB 位元組

daft.File 僅提供引用資訊，真正消費位元組時需要根據 path、offset、length 從 OSS 拉取。

def fetch_blob(file_ref: daft.File, table) -> bytes:
    """讀 BLOB 引用對應的位元組。table 是 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)

# 用法
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

步驟 6：打 tag 與時間旅行

將“訓練集 v1”打 tag 固化，便於後續按 tag 回溯到該時間點的資料。

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)

# 回到那個時點
df_v1 = read_paimon("default.drive_frames", CATALOG_OPTIONS, tag_name="training_v1")
print("training_v1 frames:", df_v1.count_rows())

也支援使用 read_paimon(..., snapshot_id=123) 按 snapshot id 進行時間旅行。

步驟 7：接入 ML 預先處理 UDF

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)              # 走 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

注意事項

• BLOB 表 Schema 限制：含 large_binary 列的表必須設定 row-tracking.enabled=true 與 data-evolution.enabled=true，否則 DLF 服務端將拒絕建表。

• 分區鍵選擇：分區鍵需與典型查詢 filter 對齊（如按 channel、按 capture_date 等），分區裁剪才能真正生效。

• BLOB 按需消費：daft.File 是 zero-copy 引用，不會觸發 OSS 讀取；僅在調用 table.file_io.new_input_stream(...) 時才會下載實際資料。