Use Daft with DLF Iceberg tables - Data Lake Formation - Alibaba Cloud Documentation Center

Daft is a high-performance distributed DataFrame engine. You can use Daft to read and write Apache Iceberg tables in Alibaba Cloud Data Lake Formation (DLF), and perform DataFrame operations such as filtering and aggregation.

Prerequisites

Note

The DLF Iceberg REST service is accessible only from within a VPC. Run the code in this topic from a VPC environment in the same region as DLF (such as an ECS instance or EMR cluster). For the endpoint of each region, see Iceberg REST endpoints.

Install dependencies

Python 3.10 or later.

Install the DLF-compatible PyIceberg package (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]"

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

Configure parameters

Prepare the following information:

Parameter	Description
`${accessKeyId}`	Your AccessKey ID.
`${accessKeySecret}`	Your AccessKey secret.
`${regionId}`	Region ID where DLF is deployed, for example, `ap-southeast-1`. For the list of region IDs, see Endpoints and public network access.
`${catalogName}`	Catalog name in DLF (corresponds to the Iceberg warehouse).
`${database}`	Name of the target database (Iceberg namespace).

Important

Protect your AccessKey credentials. Do not hardcode them in your code or commit them to code repositories. We recommend retrieving them from environment variables or a secret management service.

Connection and initialization

Connect to DLF catalog

Connect to the DLF catalog via the Iceberg REST protocol by using PyIceberg's load_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",
    },
)

Create or load a table

Create a new table

Create an Iceberg table by using PyIceberg. DLF manages the table metadata.

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

Load an existing table

To work with an existing table, call catalog.load_table().

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

Data operations

Write data

Build a DataFrame by using Daft and write it to an Iceberg table. write_iceberg returns a summary table of the data files written in this operation.

Note

Daft 0.7.15 automatically configures OSS access for Iceberg tables on oss:// paths, so write_iceberg and read_iceberg require no manual IOConfig.

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

The mode parameter supports "append" (append data) and "overwrite" (overwrite existing data).

Read data

After writing data, reload the table to get the latest snapshot and refresh the STS credentials, then read it by using Daft.

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

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

daft.read_iceberg is lazy: it returns a DataFrame handle and builds an execution plan. Data is read from OSS only when you call actions such as show() or collect().

Transform data

With the DataFrame read from the table (containing the five sample rows), you can perform filtering, derived column creation, aggregation, and sorting.

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

Complete example

The following code combines all the preceding steps and can be copied and run directly.

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

The expected output of the write step (the write_iceberg result table) looks like the following:

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

Appendix

Terminology

Component	Description
DLF	Data Lake Formation. Provides Iceberg REST Catalog services and unified table metadata management.
PyIceberg	Python client for Apache Iceberg. Connects to DLF catalogs, creates and drops tables, and commits transactions.
Daft	Distributed DataFrame engine. Writes data to and reads data from Iceberg tables, and performs DataFrame transformations.
OSS	Object Storage Service. Physical storage layer for Iceberg table data files (Parquet format).

Architecture

Daft and PyIceberg work together with a clear separation of concerns:

Control plane (PyIceberg): Connects to DLF using the Iceberg REST protocol. Handles catalog connections, table creation and deletion, and snapshot commits.
Data plane (Daft): Iceberg table data files are stored on OSS. Daft reads and writes these Parquet files in parallel and provides DataFrame computation capabilities (filtering, derived columns, aggregation, sorting, etc.).

For writes, Daft writes Parquet data files and commits Iceberg snapshots atomically through PyIceberg. For reads, Daft leverages Iceberg metadata for partition pruning and file filtering.