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

Daft reads and writes Paimon tables managed by a DLF Catalog through pypaimon's `read_paimon` / `write_paimon` functions, with predicate and projection pushdown and time-travel queries. This topic provides a minimal end-to-end example.

Note

To work with BLOB / list[binary] multimodal columns or run multimodal ETL, see Manage multimodal autonomous driving data on DLF Paimon with Daft.

Prerequisites

Step 1: Install dependencies

Download the pypaimon offline installation package: pypaimon-1.5.dev20260608.tar.gz
Install Python dependencies

In the directory that contains the offline package, run the following command to install all dependencies:
```
pip install daft pyarrow pandas requests pypaimon-1.5.dev20260608.tar.gz
```

Verify the environment

Run the following code to verify that the environment is set up correctly:

import daft
from pypaimon.daft import read_paimon, write_paimon
print("daft:", daft.__version__, "pypaimon.daft: OK")

Step 2: Configure the catalog connection

Set the catalog connection parameters as follows:

CATALOG_OPTIONS = {
    "metastore": "rest",
    "uri": "http://<DLF-ENDPOINT>",       # HTTPS is required over the public internet
    "warehouse": "<YOUR-CATALOG>",
    "token.provider": "dlf",
    "dlf.region": "<REGION-ID>",
    "dlf.access-key-id": "<ACCESS-KEY-ID>",
    "dlf.access-key-secret": "<ACCESS-KEY-SECRET>",
    "dlf.oss-endpoint": "<OSS-ENDPOINT>", # Optional for VPC; required for the public internet
}

Parameter	Description
metastore	Must be `rest`.
uri	DLF Paimon REST endpoint. For details, see Endpoints and public network access. VPC access supports both HTTP and HTTPS; public internet access requires HTTPS. Example of VPC access: http://ap-southeast-1-vpc.dlf.aliyuncs.com
warehouse	Name of the DLF catalog.
token.provider	Set to `dlf` for AccessKey authentication.
dlf.region	DLF region ID, for example `cn-hangzhou`. For details, see Endpoints and public network access.
dlf.access-key-id / dlf.access-key-secret	AccessKey pair for DLF authentication.
dlf.security-token	(Optional) STS security token.
dlf.oss-endpoint	(Optional) OSS public endpoint.

Write data

Create a new table

Before you can read or write a table with Daft, create it by using PyPaimon's create_table():

from datetime import datetime
import pyarrow as pa
from pypaimon import CatalogFactory, Schema

catalog = CatalogFactory.create(CATALOG_OPTIONS)
catalog.create_database("default", True)

table_name = "default.test_daft_paimon_" + datetime.now().strftime("%Y%m%d_%H%M%S")
pa_schema = pa.schema([
    pa.field("id", pa.int32(), nullable=False),
    pa.field("name", pa.string()),
    pa.field("score", pa.int64()),
])
schema = Schema.from_pyarrow_schema(
    pa_schema,
    options={"bucket": "-1", "file.format": "parquet"},
)
catalog.create_table(table_name, schema, True)

Write to a table

Use write_paimon to write a Daft DataFrame to an existing Paimon table:

import daft
from pypaimon.daft import write_paimon

append_df = daft.from_pydict({
    "id":    [4, 5],
    "name":  ["Dan", "Eve"],
    "score": [77, 92],
})
write_paimon(append_df, "default.your_table", CATALOG_OPTIONS, mode="append")

The mode parameter accepts two values:

"append": appends rows to the table.
"overwrite": replaces all existing data in the table.

Read an existing table

Basic read

Call read_paimon to read a table into a Daft DataFrame:

from pypaimon.daft import read_paimon

df = read_paimon("default.your_table", CATALOG_OPTIONS)
df.show()

Predicate and projection pushdown

Daft pushes filter predicates and column projections down to the Paimon scan:

import daft

df = (
    read_paimon("default.your_table", CATALOG_OPTIONS)
    .where(daft.col("score") >= 88)
    .select("id", "name")
)
df.show()

Time travel

Read a historical version by passing snapshot_id or tag_name:

# By snapshot_id
df = read_paimon("default.your_table", CATALOG_OPTIONS, snapshot_id=42)

# By tag_name
df = read_paimon("default.your_table", CATALOG_OPTIONS, tag_name="v1")

Complete example

The following example demonstrates the complete workflow: create a table, write data, append rows, read back, and apply filter and projection pushdown.

from __future__ import annotations

import os
from datetime import datetime

import daft
import pyarrow as pa
from pypaimon import CatalogFactory, Schema
from pypaimon.daft import read_paimon, write_paimon


CATALOG_OPTIONS = {
    "metastore": "rest",
    "uri": "http://<DLF-ENDPOINT>",
    "warehouse": "<YOUR-CATALOG>",
    "token.provider": "dlf",
    "dlf.region": "<REGION-ID>",
    "dlf.access-key-id": os.environ["DLF_ACCESS_KEY_ID"],
    "dlf.access-key-secret": os.environ["DLF_ACCESS_KEY_SECRET"],
    "dlf.oss-endpoint": "<OSS-ENDPOINT>",
}

DATABASE = "default"


def main() -> None:
    catalog = CatalogFactory.create(CATALOG_OPTIONS)
    catalog.create_database(DATABASE, True)

    # 1. Create the table (append table + unaware bucket)
    table_name = f"{DATABASE}.test_daft_paimon_" + datetime.now().strftime("%Y%m%d_%H%M%S")
    pa_schema = pa.schema([
        pa.field("id", pa.int32(), nullable=False),
        pa.field("name", pa.string()),
        pa.field("score", pa.int64()),
    ])
    schema = Schema.from_pyarrow_schema(
        pa_schema,
        options={"bucket": "-1", "file.format": "parquet"},
    )
    catalog.create_table(table_name, schema, True)
    print("created:", table_name)

    # 2. Write the initial batch
    initial_df = daft.from_pydict({
        "id":    [1, 2, 3],
        "name":  ["Alice", "Bob", "Carol"],
        "score": [90, 85, 88],
    })
    write_paimon(initial_df, table_name, CATALOG_OPTIONS, mode="append")

    # 3. Append more rows
    append_df = daft.from_pydict({
        "id":    [4, 5],
        "name":  ["Dan", "Eve"],
        "score": [77, 92],
    })
    write_paimon(append_df, table_name, CATALOG_OPTIONS, mode="append")

    # 4. Read the table back
    df = read_paimon(table_name, CATALOG_OPTIONS).sort("id")
    df.show()
    assert df.to_pydict()["id"] == [1, 2, 3, 4, 5]

    # 5. Apply predicate and projection pushdown
    filtered = (
        read_paimon(table_name, CATALOG_OPTIONS)
        .where(daft.col("score") >= 88)
        .select("id", "name")
        .sort("id")
    )
    filtered.show()
    assert filtered.to_pydict()["id"] == [1, 3, 5]

    print("daft + dlf + paimon: ok")


if __name__ == "__main__":
    main()

Expected output:

created: default.test_daft_paimon_20260524_xxxxxx
╭───────┬────────┬───────╮
│ id    ┆ name   ┆ score │
│ Int32 ┆ String ┆ Int64 │
╞═══════╪════════╪═══════╡
│ 1     ┆ Alice  ┆ 90    │
│ 2     ┆ Bob    ┆ 85    │
│ 3     ┆ Carol  ┆ 88    │
│ 4     ┆ Dan    ┆ 77    │
│ 5     ┆ Eve    ┆ 92    │
╰───────┴────────┴───────╯

╭───────┬────────╮
│ id    ┆ name   │
│ Int32 ┆ String │
╞═══════╪════════╡
│ 1     ┆ Alice  │
│ 3     ┆ Carol  │
│ 5     ┆ Eve    │
╰───────┴────────╯

daft + dlf + paimon: ok