数据湖构建：PyIceberg访问 DLF

环境准备与安装

PyIceberg 需要特定的 Python 环境与依赖库支持。请严格按照以下步骤进行配置，以确保与 DLF 服务的兼容性。

下载安装包

由于标准 PyPI 源中的 PyIceberg 尚未完全适配 DLF，请务必使用以下指定的开发版本：

pyiceberg-0.10.0.dev1.tar.gz

安装步骤

推荐在 Python 虚拟环境中进行安装，以隔离系统依赖。

Python版本需要3.10及以上版本

# 1. 创建并激活虚拟环境
python3 -m venv myenv
source myenv/bin/activate
# 2. 安装指定版本的 PyIceberg
# 请确保安装包位于当前目录，或替换为实际路径
pip install pyiceberg-0.10.0.dev0.tar.gz 
# 3. 安装依赖库
pip install "pyarrow>=19.0.0, <22.0.0"
pip install "boto3>=1.24.59"
pip install pandas

代码示例

以下 Python 脚本演示了完整的生命周期操作：连接 Catalog、创建表、写入数据、读取数据以及删除表。

请将脚本中的占位符替换为实际配置。

• ${regionId}：DLF域名，如cn-hangzhou。详情请参见Iceberg REST服务接入点。

• ${catalogName}：DLF Catalog 名称。

• ${accessKey}：AccessKey ID。

• ${accessKeySecret}：AccessKey Secret。

import pyarrow as pa
from pyiceberg.catalog import load_catalog
from pyiceberg.exceptions import TableAlreadyExistsError, NoSuchTableError
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.transforms import IdentityTransform
from pyiceberg.types import LongType, NestedField
# 1. 配置与连接 Catalog
catalog = load_catalog(
    "default",
    **{
        "type": "rest",
        "uri": "http://${regionId}-vpc.dlf.aliyuncs.com/iceberg",
        "warehouse": "${catalogName}",
        "rest.signing-name": "DlfNext",
        "rest.signing-region": "${regionId}",
        "rest.sigv4-enabled": "true",
        "client.access-key-id": "${accessKey}",
        "client.secret-access-key": "{accessKeyId}",
        "client.region": "${regionId}",
    },
)
# ---------------------------------------------------
# 2. 定义表结构与元数据
# ---------------------------------------------------
TEST_TABLE_SCHEMA = Schema(
    NestedField(1, "x", LongType(), required=True),
    NestedField(2, "y", LongType(), doc="comment", required=True),
    NestedField(3, "z", LongType(), required=True),
)
TEST_TABLE_IDENTIFIER = ("default", "my_table")
TEST_TABLE_PARTITION_SPEC = PartitionSpec(
    PartitionField(name="x", transform=IdentityTransform(), source_id=1, field_id=1000)
)
TEST_TABLE_PROPERTIES = {"read.split.target.size": "134217728"}  # 128MB
# ---------------------------------------------------
# 3. 执行测试流程
# ---------------------------------------------------
# 清理旧表（如果存在）
try:
    catalog.drop_table(identifier=TEST_TABLE_IDENTIFIER)
    print("Existing table dropped.")
except NoSuchTableError:
    print("No existing table to drop.")
# 创建新表
try:
    catalog.create_table(
        identifier=TEST_TABLE_IDENTIFIER,
        schema=TEST_TABLE_SCHEMA,
        partition_spec=TEST_TABLE_PARTITION_SPEC,
        properties=TEST_TABLE_PROPERTIES,
    )
    print("Table created.")
except TableAlreadyExistsError:
    print("Table already exists, will append data.")
# 加载表
table = catalog.load_table(identifier=TEST_TABLE_IDENTIFIER)
print(f"Loaded table: {table}")
# 构造 PyArrow 数据
arrow_schema = schema_to_pyarrow(table.schema())
data = pa.Table.from_pydict(
    {
        "x": [1, 2, 3],
        "y": [10, 20, 30],
        "z": [100, 200, 300],
    },
    schema=arrow_schema,
)
# 写入数据
print(f"Inserting {data.num_rows} rows...")
table.append(data)
print("Insert finished.")
# 读取并展示数据
scan = table.scan()
df = scan.to_pandas()
print("First 10 rows via to_pandas():")
print(df.head(10))
# 清理测试表
try:
    catalog.drop_table(identifier=TEST_TABLE_IDENTIFIER)
    print("Table dropped after test.")
except NoSuchTableError:
    print("Table already dropped.")

运行结果展示：

(myenv) root@iZbp1h4zr65vjcvz9w3080Z:~/workspace# python test.py
Existing table dropped.
Table created.
Loaded table: my_table(
    1: x: required long,
    2: y: required long (comment),
    3: z: required long
),
partition by: [x],
sort order: [],
snapshot: null
Inserting 3 rows...
Insert finished.
First 10 rows via to_pandas():
   x   y    z
0  1  10  100
1  2  20  200
2  3  30  300
Table dropped after test.
(myenv) root@iZbp1h4zr65vjcvz9w3080Z:~/workspace# pip list|grep -E "pyarrow|pyiceberg|boto3|pandas"
boto3                 1.42.15
pandas                2.3.3
pyarrow               19.0.0
pyiceberg             0.10.0.dev0
(myenv) root@iZbp1h4zr65vjcvz9w3080Z:~/workspace#