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

Daft是一款高性能的分布式DataFrame引擎。本文介绍如何使用Daft读写DLF Iceberg表，并进行数据过滤、聚合等变换操作。

环境准备

说明

DLF Iceberg REST服务仅支持VPC内网访问，因此需在与DLF同地域的VPC环境（如ECS、EMR集群）中运行本文代码。各地域的VPC Endpoint请参见Iceberg REST服务接入点。

安装依赖

Python 3.10及以上版本。

安装DLF适配版PyIceberg（pyiceberg-dlf）。

python3 -m venv venv
source venv/bin/activate
pip install -U pip
# 卸载官方版（与 DLF 适配版不能共存）
pip uninstall -y pyiceberg
# rest-sigv4 为必选项（安装 boto3，用于 REST sigv4 签名）
pip install "pyiceberg-dlf[rest-sigv4,pyarrow,pandas]"

pyiceberg-dlf说明详见PyIceberg访问 DLF：环境准备与安装。

安装Daft。
```
pip install "daft>=0.7.17"
```

配置参数

准备以下信息：

参数	说明
`${accessKeyId}`	阿里云账号的AccessKey ID。
`${accessKeySecret}`	阿里云账号的AccessKey Secret。
`${regionId}`	DLF所在地域的ID，例如`cn-hangzhou`。各地域ID请参见服务接入点与公网访问。
`${catalogName}`	DLF中的Catalog名称（对应Iceberg warehouse）。
`${database}`	目标数据库（对应Iceberg namespace）名称。

重要

请妥善保管AccessKey，避免硬编码到代码或提交到代码仓库，建议通过环境变量或密钥管理服务读取。

连接与初始化

连接 DLF Catalog

通过Iceberg REST协议，使用PyIceberg的load_catalog连接DLF 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",
    },
)

创建或加载表

创建新表

通过PyIceberg创建一张Iceberg表，表元数据由DLF管理。

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

加载已有表

如需操作已有表，使用catalog.load_table()。

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

数据操作

写入数据

使用Daft构建DataFrame并写入Iceberg表。write_iceberg返回一张汇总本次写入数据文件的结果表。

说明

Daft 0.7.15+对oss://路径的Iceberg表自动配置OSS访问，write_iceberg和read_iceberg均无需手动构建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()

mode支持"append"（追加写入）和"overwrite"（覆盖写入）。

读取数据

写入后，重新加载表以获取最新快照（同时刷新STS凭证），再用Daft读取。

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

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

daft.read_iceberg是惰性的：它返回一个DataFrame句柄并构建执行计划，只有在调用show() / collect()等动作时才真正从OSS读取数据。

数据变换

基于读取的DataFrame（包含5行示例数据），可以进行过滤、派生列、聚合、排序等操作。

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

# 过滤：value大于30的行
df.where(df["value"] > 30).show()

# 派生列：新增value_x2 = value * 2
df.with_column("value_x2", df["value"] * 2).show()

# 分组聚合：按category统计行数与value之和
df.groupby("category").agg(
    daft.col("id").count().alias("row_count"),
    daft.col("value").sum().alias("value_sum"),
).sort("category").show()

# 排序：按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

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


def create_catalog():
    """连接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):
    """通过PyIceberg创建一张未分区的Iceberg表。"""
    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. 创建表（PyIceberg）
        table = create_table(catalog, table_name)
        print("表已创建，数据位置：", table.location())

        # 2. 使用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. 重新加载表以获取最新快照，使用Daft读取
        table = catalog.load_table((DATABASE, table_name))
        result = daft.read_iceberg(table).collect()
        result.show()

        # 4. DataFrame变换
        result.groupby("category").agg(
            daft.col("id").count().alias("row_count"),
            daft.col("value").sum().alias("value_sum"),
        ).sort("category").show()
    finally:
        # 5. 删除表（PyIceberg）
        if table is not None:
            catalog.drop_table((DATABASE, table_name))
            print("表已删除：", table_name)


if __name__ == "__main__":
    main()

写入步骤的预期输出（write_iceberg返回的结果表）形如：

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

附录

术语说明

组件	说明
DLF	阿里云数据湖构建，提供Iceberg REST Catalog服务，统一管理表元数据。
PyIceberg	Apache Iceberg的Python客户端。本文用于连接DLF Catalog、创建/删除表、提交事务。
Daft	分布式DataFrame引擎。本文用于向Iceberg表写入数据、读取数据并做DataFrame变换。
OSS	阿里云对象存储，Iceberg表数据文件（Parquet）的物理存储。

技术架构

Daft与PyIceberg分工协作：

控制面（PyIceberg）：通过Iceberg REST协议连接DLF，完成Catalog连接、建表、删表、快照提交。
数据面（Daft）：Iceberg表的数据文件存储在OSS上，Daft直接并行读写这些Parquet文件，并提供DataFrame计算能力（过滤、派生列、聚合、排序等）。

写入时，Daft写出Parquet数据文件，并通过PyIceberg原子提交Iceberg快照；读取时，Daft借助Iceberg元数据做分区裁剪与文件过滤。