Data Lake Formation:Utiliser Daft avec les tables DLF Iceberg

Installer les dépendances

1. Python 3.10 ou version ultérieure.

2. 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]"

3. Installez Daft.

   pip install "daft>=0.7.17"

Configurer les paramètres

Préparez les informations suivantes :

Se connecter au catalog DLF

Connectez-vous au catalog 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",
    },
)

Créer ou charger une table

from pyiceberg.schema import Schema
from pyiceberg.types import LongType, StringType, DoubleType, NestedField
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC

schema = Schema(
    NestedField(field_id=1, name="id", field_type=LongType(), required=True),
    NestedField(field_id=2, name="name", field_type=StringType(), required=False),
    NestedField(field_id=3, name="category", field_type=StringType(), required=False),
    NestedField(field_id=4, name="value", field_type=DoubleType(), required=False),
)

table = catalog.create_table(
    identifier=("${database}", "daft_demo_table"),
    schema=schema,
    partition_spec=UNPARTITIONED_PARTITION_SPEC,
)
print("Table created. Data location:", table.location())

table = catalog.load_table(("${database}", "table_name"))

Écrire des données

Construisez un DataFrame à l'aide de Daft et écrivez-le dans une table Iceberg. La fonction write_iceberg renvoie un tableau récapitulatif des fichiers de données écrits lors de cette opération.

import daft

df = daft.from_pydict({
    "id": [1, 2, 3, 4, 5],
    "name": ["item_1", "item_2", "item_3", "item_4", "item_5"],
    "category": ["A", "B", "A", "B", "A"],
    "value": [10.5, 21.0, 31.5, 42.0, 52.5],
})

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

Le paramètre mode prend en charge les valeurs "append" (ajout de données) et "overwrite" (écrasement des données existantes).

Lire des données

Après avoir écrit des données, rechargez la table pour obtenir le dernier snapshot et actualiser les identifiants STS, puis lisez-la à l'aide de Daft.

table = catalog.load_table(("${database}", "daft_demo_table"))

df = daft.read_iceberg(table)
df.show()

La fonction daft.read_iceberg est lazy : elle renvoie un handle DataFrame et construit un plan d'exécution. Les données sont lues depuis OSS uniquement lorsque vous appelez des actions telles que show() ou collect().

Transformer des données

À partir du DataFrame lu depuis la table (contenant les cinq lignes d'exemple), vous pouvez filtrer les données, créer des colonnes dérivées, agréger et trier.

df = daft.read_iceberg(table).collect()

# Filter: rows where value is greater than 30
df.where(df["value"] > 30).show()

# Derived column: add value_x2 = value * 2
df.with_column("value_x2", df["value"] * 2).show()

# Group and aggregate: count rows and sum values by category
df.groupby("category").agg(
    daft.col("id").count().alias("row_count"),
    daft.col("value").sum().alias("value_sum"),
).sort("category").show()

# Sort: descending by value
df.sort("value", desc=True).show()

import uuid

import daft
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import LongType, StringType, DoubleType, NestedField
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC

# ==================== Configuration ====================
ACCESS_KEY_ID = "${accessKeyId}"
ACCESS_KEY_SECRET = "${accessKeySecret}"
REGION = "${regionId}"
CATALOG_NAME = "${catalogName}"
DATABASE = "${database}"
# =======================================================

def create_catalog():
    """Connect to the DLF Iceberg REST Catalog."""
    return load_catalog(
        "dlf",
        **{
            "type": "rest",
            "uri": f"http://{REGION}-vpc.dlf.aliyuncs.com/iceberg",
            "warehouse": CATALOG_NAME,
            "rest.signing-name": "DlfNext",
            "rest.signing-region": REGION,
            "rest.sigv4-enabled": "true",
            "client.access-key-id": ACCESS_KEY_ID,
            "client.secret-access-key": ACCESS_KEY_SECRET,
            "client.region": REGION,
            "s3.endpoint": f"https://oss-{REGION}-internal.aliyuncs.com",
        },
    )

def create_table(catalog, table_name):
    """Create an unpartitioned Iceberg table using PyIceberg."""
    schema = Schema(
        NestedField(field_id=1, name="id", field_type=LongType(), required=True),
        NestedField(field_id=2, name="name", field_type=StringType(), required=False),
        NestedField(field_id=3, name="category", field_type=StringType(), required=False),
        NestedField(field_id=4, name="value", field_type=DoubleType(), required=False),
    )
    return catalog.create_table(
        identifier=(DATABASE, table_name),
        schema=schema,
        partition_spec=UNPARTITIONED_PARTITION_SPEC,
    )

def main():
    catalog = create_catalog()
    table_name = f"daft_demo_{uuid.uuid4().hex[:8]}"
    table = None
    try:
        # 1. Create table (PyIceberg)
        table = create_table(catalog, table_name)
        print("Table created. Data location:", table.location())

        # 2. Write data using Daft
        df = daft.from_pydict({
            "id": [1, 2, 3, 4, 5],
            "name": ["item_1", "item_2", "item_3", "item_4", "item_5"],
            "category": ["A", "B", "A", "B", "A"],
            "value": [10.5, 21.0, 31.5, 42.0, 52.5],
        })
        df.write_iceberg(table, mode="append").show()

        # 3. Reload table to get latest snapshot, then read using Daft
        table = catalog.load_table((DATABASE, table_name))
        result = daft.read_iceberg(table).collect()
        result.show()

        # 4. DataFrame transformations
        result.groupby("category").agg(
            daft.col("id").count().alias("row_count"),
            daft.col("value").sum().alias("value_sum"),
        ).sort("category").show()
    finally:
        # 5. Drop table (PyIceberg)
        if table is not None:
            catalog.drop_table((DATABASE, table_name))
            print("Table dropped:", table_name)

if __name__ == "__main__":
    main()

╭───────────┬───────┬───────────┬────────────────────────────────╮
│ operation ┆ rows  ┆ file_size ┆ file_name                      │
╞═══════════╪═══════╪═══════════╪════════════════════════════════╡
│ ADD       ┆ 5     ┆ 1708      ┆ oss://<bucket>/.../xxx.parquet │
╰───────────┴───────┴───────────┴────────────────────────────────╯

Terminologie

Architecture

Daft et PyIceberg travaillent conjointement avec une séparation claire des responsabilités :

• Plan de contrôle (PyIceberg) : se connecte à DLF via le protocole REST Iceberg. Gère les connexions au catalog, la création et la suppression de tables, ainsi que la validation des snapshots.

• Plan de données (Daft) : les fichiers de données des tables Iceberg sont stockés sur OSS. Daft lit et écrit ces fichiers Parquet en parallèle et fournit des capacités de calcul DataFrame (filtrage, colonnes dérivées, agrégation, tri, etc.).

Pour les écritures, Daft écrit les fichiers de données Parquet et valide les snapshots Iceberg de manière atomique via PyIceberg. Pour les lectures, Daft exploite les métadonnées Iceberg pour l'élimination des partitions et le filtrage des fichiers.