使用Daft读写多模态Iceberg数据 - 数据湖构建

本文介绍如何将图像字节直接存入DLF Iceberg表，并使用Daft进行缩略图生成、embedding计算和视觉相似度检索。

多模态存储模式简介

本文采用内联存储模式：将图像或音视频的原始字节、缩略图和embedding向量与元数据存放在同一张Iceberg表中。Daft可以直接从表内解码图像字节，无需再通过对象存储路径取数，写入、读取和相似度检索在一张表内即可完成。

数据	列类型
原图字节	`binary`
缩略图（可选，便于轻量浏览）	`binary`
embedding向量	`list<float>`

说明

单条媒体文件较大时，内联存储会增大Parquet文件体积，请根据数据规模权衡选择。

环境准备

安装依赖

Python 3.10及以上版本。

安装DLF适配版PyIceberg（pyiceberg-dlf）。

python3 -m venv venv
source venv/bin/activate
pip install -U pip
# 卸载官方版（与 DLF 适配版不能共存）
pip uninstall -y pyiceberg
# rest-sigv4 为必选项（安装 boto3，用于 REST sigv4 签名）
pip install "pyiceberg-dlf[rest-sigv4,pyarrow,pandas]"

pyiceberg-dlf说明详见PyIceberg访问 DLF：环境准备与安装。

安装Daft。
```
pip install "daft>=0.7.17"
```

配置参数

准备以下信息：

参数	说明
`${accessKeyId}`	阿里云账号的AccessKey ID。
`${accessKeySecret}`	阿里云账号的AccessKey Secret。
`${regionId}`	DLF所在地域的ID，例如`cn-hangzhou`。各地域ID请参见服务接入点与公网访问。
`${catalogName}`	DLF中的Catalog名称（对应Iceberg warehouse）。
`${database}`	目标数据库（对应Iceberg namespace）名称。

重要

请妥善保管AccessKey，避免硬编码到代码或提交到代码仓库，建议通过环境变量或密钥管理服务读取。

连接 DLF Catalog

通过Iceberg REST协议，使用PyIceberg的load_catalog连接DLF Catalog。

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

准备样例图片

准备两张本地JPEG样例图片n01.jpg、n02.jpg（RGB格式），放到当前目录。可从公开数据集Imagenette（https://github.com/fastai/imagenette）的n01440764（tench）和n02979186（cassette player）类别中各取一张并重命名。

创建多模态表

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

说明

以上建表语句默认创建format-version 2表。DLF支持创建format-version 3表，但当前PyIceberg和Daft尚不支持写入v3表和VARIANT类型，多模态表请勿通过properties={"format-version": "3"}指定v3。

写入图像数据

将原图字节放入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列的内联字节，无需再访问对象存储。

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与相似度检索

使用@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)))

获取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))

说明

示例中的embedding为降采样描述子，仅用于演示。实际使用时可替换为CLIP、ResNet等模型推理。