使用Daft操作DLF Iceberg表 - Data Lake Formation

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，避免寫入程式碼到代碼或提交到代碼倉庫，建議通過環境變數或Key Management Service讀取。

串連與初始化

串連 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	阿里雲Object Storage Service，Iceberg表資料檔案（Parquet）的實體儲存體。

技術架構

Daft與PyIceberg分工協作：

控制面（PyIceberg）：通過Iceberg REST協議串連DLF，完成Catalog串連、建表、刪表、快照提交。
資料面（Daft）：Iceberg表的資料檔案儲存在OSS上，Daft直接並行讀寫這些Parquet檔案，並提供DataFrame計算能力（過濾、衍生的資料行、彙總、排序等）。

寫入時，Daft寫出Parquet資料檔案，並通過PyIceberg原子提交Iceberg快照；讀取時，Daft藉助Iceberg中繼資料做分區裁剪與檔案過濾。