使用 Daft 操作 DLF Paimon 表 - 数据湖构建

Daft 通过 pypaimon 的 read_paimon / write_paimon 接口读写 DLF Catalog 管理的 Paimon 表，支持谓词与列裁剪下推、时间旅行查询。本文给出最小可运行的端到端示例。

说明

如需操作 BLOB / list[binary] 等多模态列或执行多模态 ETL，请参见基于Daft和DLF的自动驾驶多模态数据湖实践。

前提条件

步骤一：安装依赖

下载pypaimon离线安装包pypaimon-1.5.dev20260608.tar.gz
安装 Python 依赖
在pypaimon离线安装包所在目录执行以下命令安装所有依赖：
```
pip install daft pyarrow pandas requests pypaimon-1.5.dev20260608.tar.gz
```

验证环境

执行以下代码验证环境是否就绪：

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

步骤二：配置 Catalog 连接

通过以下配置连接 DLF Catalog：

CATALOG_OPTIONS = {
    "metastore": "rest",
    "uri": "http://<DLF-ENDPOINT>",       # 公网访问须使用 HTTPS
    "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>", # VPC 可选，公网必填
}

参数说明如下：

参数	说明
metastore	固定为 rest。
uri	DLF Paimon REST 访问端点，详见服务接入点与公网访问。VPC 访问支持 HTTP 和 HTTPS 协议，公网访问必须使用 HTTPS 协议。VPC访问示例：http://ap-southeast-1-vpc.dlf.aliyuncs.com
warehouse	DLF Catalog 名称。
token.provider	AccessKey 认证模式下填 dlf。
dlf.region	DLF 所属地域 ID（如 cn-hangzhou），详见服务接入点与公网访问。
dlf.access-key-id / dlf.access-key-secret	DLF AccessKey。
dlf.security-token	（可选）STS 安全令牌。
dlf.oss-endpoint	（可选）OSS 公网访问 endpoint。

写入数据

创建新表

新表必须先用 PyPaimon 的 create_table() 创建，再用 Daft 读写：

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_paimon 将 Daft DataFrame 写入已有 Paimon 表：

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

mode 支持以下两种取值：

"append"：追加写入。
"overwrite"：覆盖写入。

读取已有表

基本读取

使用 read_paimon 读取指定表的全部数据：

from pypaimon.daft import read_paimon

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

谓词与列裁剪

谓词与列裁剪会自动下推到 Paimon 扫描层：

import daft

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

时间旅行

支持通过 snapshot_id 或 tag_name 读取历史版本：

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

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

完整示例

以下示例演示了完整的端到端流程：创建新表 → Daft 写入 → 追加 → 读取 → 过滤投影。

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. 建表（append 表 + 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. Daft 写入首批数据
    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. Daft 追加
    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. Daft 读取
    df = read_paimon(table_name, CATALOG_OPTIONS).sort("id")
    df.show()
    assert df.to_pydict()["id"] == [1, 2, 3, 4, 5]

    # 5. 谓词 + 投影下推
    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()

预期输出如下：

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