Utiliser Daft pour lire et écrire des données Iceberg multimodales - Data Lake Formation

Aperçu du stockage multimodal

Avec l'approche de stockage en ligne, les octets bruts d'images ou de fichiers audio/vidéo, les vignettes et les vecteurs d'embedding sont stockés aux côtés des métadonnées dans la même table Iceberg. Daft décode les octets d'image directement depuis la table sans accéder aux chemins de stockage d'objets : les écritures, les lectures et les recherches de similarité s'effectuent sur une seule table.

Données

Type de colonne

Octets d'image bruts

binary

Vignette (facultative ; pour une navigation légère)

binary

Vecteur d'embedding

list<float>

Remarque

Lorsque les fichiers multimédias individuels sont volumineux, le stockage en ligne augmente la taille des fichiers Parquet. Choisissez votre approche de stockage en fonction du volume de vos données.

Configuration de l'environnement

Installer les dépendances

Python 3.10 ou version ultérieure.

Installez le package PyIceberg compatible avec DLF (pyiceberg-dlf).

python3 -m venv venv
source venv/bin/activate
pip install -U pip
# Uninstall pyiceberg, which cannot coexist with pyiceberg-dlf
pip uninstall -y pyiceberg
# rest-sigv4 is required (installs boto3 for REST sigv4 signing)
pip install "pyiceberg-dlf[rest-sigv4,pyarrow,pandas]"

Installez Daft.
```
pip install "daft>=0.7.17"
```

Configurer les paramètres

Préparez les informations suivantes :

Paramètre

Description

${accessKeyId}

Votre ID AccessKey.

${accessKeySecret}

Votre secret AccessKey.

${regionId}

ID de la région où DLF est déployé, par exemple ap-southeast-1. Pour la liste des ID de région, consultez Régions et endpoints.

${catalogName}

Nom du catalogue dans DLF (correspond à l'entrepôt Iceberg).

${database}

Nom de la base de données cible (namespace Iceberg).

Important

Protégez vos identifiants AccessKey. Ne les codez pas en dur dans votre code et ne les validez pas dans les dépôts de code. Nous vous recommandons de les récupérer à partir de variables d'environnement ou d'un service de gestion des secrets.

Se connecter au catalogue DLF

Connectez-vous au catalogue DLF via le protocole REST Iceberg en utilisant la fonction load_catalog de PyIceberg.

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

Préparer des images d'exemple

Préparez deux exemples d'images JPEG locales nommées n01.jpg et n02.jpg (format RGB) et placez-les dans le répertoire courant. Vous pouvez télécharger une image de chaque classe n01440764 (tench) et n02979186 (cassette player) à partir du jeu de données public Imagenette et les renommer.

Créer une table multimodale

Utilisez PyIceberg pour créer une table multimodale avec des colonnes pour les octets d'image et les vecteurs d'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,
)

Remarque

Cette instruction crée par défaut une table de format-version 2. DLF prend en charge les tables de format-version 3, mais PyIceberg et Daft ne prennent pas encore en charge l'écriture dans les tables v3 ni le type VARIANT. Ne définissez pas properties={"format-version": "3"} pour les tables multimodales.

Écrire des données d'image

Stockez les octets d'image bruts dans la colonne image et utilisez les fonctions d'image de Daft pour générer des vignettes. Daft 0.7.15 et versions ultérieures configurent OSS automatiquement, il n'est donc pas nécessaire de transmettre manuellement 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()],
})

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

Lire et traiter les images

Après avoir lu la table, décodez directement les octets en ligne dans la colonne image : aucun accès au stockage d'objets n'est requis.

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

Calculer les embeddings et effectuer une recherche de similarité

Utilisez @daft.func pour convertir les images en vecteurs et produire une colonne d'embedding list<float>. L'exemple suivant utilise un descripteur sous-échantillonné ; vous pouvez le remplacer par CLIP, ResNet ou un autre modèle.

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

Une fois la colonne embedding prête, appelez collect() pour rassembler les données localement, calculer la similarité cosinus par rapport à un vecteur d'image de requête et récupérer les résultats Top-K.

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

Exemple complet

L'exemple suivant détaille le workflow de bout en bout : connexion à DLF, création d'une table, écriture des octets d'image avec vignettes et embeddings, navigation dans les données, exécution d'une recherche visuelle, décodage d'une image et nettoyage.

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

Remarque

Les embeddings de cet exemple sont des descripteurs sous-échantillonnés utilisés uniquement à des fins de démonstration. En production, remplacez-les par CLIP, ResNet ou un autre modèle pour l'inférence.