Data Lake Formation:Use Daft to read and write multimodal Iceberg data

Multimodal storage overview

In the inline storage approach, raw image or audio/video bytes, thumbnails, and embedding vectors are stored alongside metadata in the same Iceberg table. Daft decodes image bytes directly from the table without accessing object storage paths, so writes, reads, and similarity searches all operate on a single table.

Environment setup

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

Create a multimodal table

Use PyIceberg to create a multimodal table with columns for image bytes and embedding vectors:

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

Write image data

Store raw image bytes in the image column and use Daft image functions to derive thumbnails. Daft 0.7.15 and later configures OSS automatically, so you do not need to pass io_config manually.

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

# Derive thumbnails from raw image bytes: decode -> resize to 32x32 -> re-encode as 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")

Read and process images

After reading the table, decode the inline bytes in the image column directly -- no object storage access required.

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

Compute embeddings and similarity search

1. Use @daft.func to convert images into vectors and produce a list<float> embedding column. The following example uses a downsampled descriptor; you can replace it with CLIP, ResNet, or another model.

   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()
   
   # Decode inline images, resize to uniform dimensions, and compute embeddings (can also be persisted at write time; see the complete example)
   df = daft.read_iceberg(table)
   df = df.with_column("embedding", embed(image.resize(image.decode_image(df["image"]), 8, 8)))

2. With the embedding column ready, call collect() to gather the data locally, compute cosine similarity against a query image vector, and retrieve the top-K results.

   # Collect the embeddings computed in the previous step
   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"])}
   
   # Cosine similarity
   cos = lambda a, b: float(a @ b / ((np.linalg.norm(a) * np.linalg.norm(b)) or 1))
   
   # Use image_id=1 as the query image and retrieve Top-K nearest neighbors
   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)

Complete example

The following example walks through the end-to-end workflow: connect to DLF, create a table, write image bytes with thumbnails and embeddings, browse data, run a visual search, decode an image, and clean up.

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) Connect to 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"})

# Prepare sample image bytes (place two JPEG images in the current directory beforehand)
samples = [("n01.jpg", "tench", open("n01.jpg", "rb").read()),
           ("n02.jpg", "cassette_player", open("n02.jpg", "rb").read())]

# 2) Create a multimodal table
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) Write image bytes with derived thumbnails and embeddings
    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) Browse data (projection pushdown, no pixel columns)
    table = catalog.load_table((DB, name))
    daft.read_iceberg(table).select("image_id", "label", "filename").sort("image_id").show()

    # 5) Visual search: find cosine neighbors for 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) Retrieve and decode an image
    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))