Data Lake Formation：使用Daft讀寫多模態Iceberg資料

多模態儲存模式簡介

本文採用內聯儲存模式：將映像或音視頻的原始位元組、縮圖和embedding向量與中繼資料存放在同一張Iceberg表中。Daft可以直接從表內解碼映像位元組，無需再通過Object Storage Service路徑取數，寫入、讀取和相似性檢索在一張表內即可完成。

環境準備

from pyiceberg.catalog import load_catalog

REGION = "${regionId}"

catalog = load_catalog(
    "dlf",
    **{
        "type": "rest",
        "uri": f"http://{REGION}-vpc.dlf.aliyuncs.com/iceberg",
        "warehouse": "${catalogName}",
        "rest.signing-name": "DlfNext",
        "rest.signing-region": REGION,
        "rest.sigv4-enabled": "true",
        "client.access-key-id": "${accessKeyId}",
        "client.secret-access-key": "${accessKeySecret}",
        "client.region": REGION,
        "s3.endpoint": f"https://oss-{REGION}-internal.aliyuncs.com",
    },
)

建立多模態表

使用PyIceberg建立包含映像位元組和embedding向量列的多模態表。

from pyiceberg.schema import Schema
from pyiceberg.types import (
    LongType, StringType, BinaryType, ListType, FloatType, NestedField,
)
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC

schema = Schema(
    NestedField(1, "image_id", LongType(), required=True),
    NestedField(2, "filename", StringType()),
    NestedField(3, "label", StringType()),
    NestedField(4, "image", BinaryType()),
    NestedField(5, "thumbnail", BinaryType()),
    NestedField(6, "embedding",
                ListType(element_id=7, element_type=FloatType(), element_required=False)),
)
table = catalog.create_table(
    ("${database}", "image_catalog"),
    schema=schema,
    partition_spec=UNPARTITIONED_PARTITION_SPEC,
)

寫入映像資料

將原圖位元組放入image列，通過Daft的映像函數從同一位元組派生縮圖，一併寫入。Daft 0.7.15及以上版本會自動設定OSS，無需手動傳入io_config。

import daft
from daft.functions import image

df = daft.from_pydict({
    "image_id": [1, 2],
    "filename": ["n01.jpg", "n02.jpg"],
    "label": ["tench", "cassette_player"],
    "image": [open("n01.jpg", "rb").read(),
              open("n02.jpg", "rb").read()],
})

# 從原圖位元組派生縮圖：解碼 → resize 32×32 → 重新編碼為JPEG
df = df.with_column("thumbnail",
        image.encode_image(image.resize(image.decode_image(df["image"]), 32, 32), "JPEG"))

df.write_iceberg(table, mode="append")

讀取與處理映像

讀取後直接解碼image列的內聯位元組，無需再訪問Object Storage Service。

from daft.functions import image

df = daft.read_iceberg(table)
df = df.with_column("img", image.decode_image(df["image"]))
df = df.with_column("thumb", image.resize(df["img"], 64, 64))
df.show()

計算embedding與相似性檢索

1. 使用@daft.func將映像轉換為向量（以下樣本為降採樣描述子，可替換為CLIP或ResNet等模型推理），得到list<float>類型的embedding列。

   import numpy as np
   import daft
   from daft.functions import image
   
   @daft.func(return_dtype=daft.DataType.list(daft.DataType.float32()))
   def embed(img) -> list:
       v = np.asarray(img).astype("float32").reshape(-1)
       return (v / (np.linalg.norm(v) or 1.0)).tolist()
   
   # 解碼內嵌影像 → 縮放到統一尺寸 → 計算embedding（也可在寫入資料時一併持久化，見完整樣本）
   df = daft.read_iceberg(table)
   df = df.with_column("embedding", embed(image.resize(image.decode_image(df["image"]), 8, 8)))

2. 擷取embedding列後，通過collect()收集到本地，與查詢圖片的向量計算餘弦相似性並取Top-K，完成視覺檢索。

   # 收集上一步計算的embedding到本地
   d = df.select("image_id", "embedding").collect().to_pydict()
   vectors = {i: np.asarray(v, "float32") for i, v in zip(d["image_id"], d["embedding"])}
   
   # 餘弦相似性
   cos = lambda a, b: float(a @ b / ((np.linalg.norm(a) * np.linalg.norm(b)) or 1))
   
   # 以image_id=1為查詢圖片，檢索Top-K近鄰
   query_id = 1
   results = sorted(
       ((i, cos(vectors[query_id], v)) for i, v in vectors.items() if i != query_id),
       key=lambda r: -r[1],
   )
   print(results)

完整樣本

以下樣本展示完整流程：串連DLF → 建表 → 映像位元組入表並產生縮圖和embedding → 瀏覽資料 → 視覺檢索 → 解碼取圖 → 清理。

import uuid
import numpy as np
import daft
from daft.functions import image
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import (
    LongType, StringType, BinaryType, ListType, FloatType, NestedField)
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC

REGION, CATALOG, DB = "${regionId}", "${catalogName}", "${database}"

# 1) 串連DLF
catalog = load_catalog("dlf", **{
    "type": "rest", "uri": f"http://{REGION}-vpc.dlf.aliyuncs.com/iceberg",
    "warehouse": CATALOG, "rest.signing-name": "DlfNext",
    "rest.signing-region": REGION, "rest.sigv4-enabled": "true",
    "client.access-key-id": "${accessKeyId}",
    "client.secret-access-key": "${accessKeySecret}", "client.region": REGION,
    "s3.endpoint": f"https://oss-{REGION}-internal.aliyuncs.com"})

# 準備範例圖片位元組（請提前在目前的目錄放置兩張JPEG圖片）
samples = [("n01.jpg", "tench", open("n01.jpg", "rb").read()),
           ("n02.jpg", "cassette_player", open("n02.jpg", "rb").read())]

# 2) 建多模態表
name = f"image_catalog_{uuid.uuid4().hex[:8]}"
table = catalog.create_table((DB, name), Schema(
    NestedField(1, "image_id", LongType(), required=True),
    NestedField(2, "filename", StringType()),
    NestedField(3, "label", StringType()),
    NestedField(4, "image", BinaryType()),
    NestedField(5, "thumbnail", BinaryType()),
    NestedField(6, "embedding",
                ListType(element_id=7, element_type=FloatType(), element_required=False))),
    partition_spec=UNPARTITIONED_PARTITION_SPEC)

@daft.func(return_dtype=daft.DataType.list(daft.DataType.float32()))
def embed(img) -> list:
    v = np.asarray(img).astype("float32").reshape(-1)
    return (v / (np.linalg.norm(v) or 1.0)).tolist()

try:
    # 3) 原圖位元組入表，同時派生縮圖和embedding
    rows = {"image_id": [], "filename": [], "label": [], "image": []}
    for i, (fn, label, jpg) in enumerate(samples, 1):
        rows["image_id"].append(i)
        rows["filename"].append(fn)
        rows["label"].append(label)
        rows["image"].append(jpg)
    df = daft.from_pydict(rows)
    df = df.with_column("thumbnail",
            image.encode_image(image.resize(image.decode_image(df["image"]), 32, 32), "JPEG"))
    df = df.with_column("embedding",
            embed(image.resize(image.decode_image(df["image"]), 8, 8)))
    df.select("image_id", "filename", "label", "image", "thumbnail",
              "embedding").write_iceberg(table, mode="append")

    # 4) 讀回瀏覽（投影下推，不讀像素）
    table = catalog.load_table((DB, name))
    daft.read_iceberg(table).select("image_id", "label", "filename").sort("image_id").show()

    # 5) 視覺檢索：對image_id=1取餘弦近鄰
    d = daft.read_iceberg(table).select("image_id", "embedding").collect().to_pydict()
    M = {i: np.asarray(v, "float32") for i, v in zip(d["image_id"], d["embedding"])}
    cos = lambda a, b: float(a @ b / ((np.linalg.norm(a) * np.linalg.norm(b)) or 1))
    print(sorted(((i, cos(M[1], v)) for i, v in M.items() if i != 1), key=lambda r: -r[1]))

    # 6) 取出一張圖並解碼
    one = daft.read_iceberg(table).where(daft.col("image_id") == 1)
    one = one.with_column("img", image.decode_image(one["image"]))
    print("decoded shape:", one.select("img").collect().to_pydict()["img"][0].shape)
finally:
    catalog.drop_table((DB, name))