PyIceberg REST Catalog による Iceberg テーブルのクエリ - DLF - Alibaba Cloud - Data Lake Formation

このトピックでは、EMR on ECS クラスターに PyIceberg をインストールし、Iceberg REST プロトコルを使用して Data Lake Formation (DLF) にアクセスするように設定する方法を説明します。

前提条件とインストール

pyiceberg-dlf のインストール

pyiceberg-dlf は、PyPI に公開されている DLF 互換の PyIceberg ディストリビューションです。これには、REST sigv4 の修正、提供される OSS ストレージ認証情報、および有効期限が切れる前の自動的な認証情報の更新が含まれています。

要件：Python 3.10 以降。

python3 -m venv venv
source venv/bin/activate
pip install -U pip
# 公式パッケージをアンインストール (pyiceberg-dlf と共存不可)
pip uninstall -y pyiceberg
# rest-sigv4 が必要 (REST sigv4 署名のために boto3 をインストール)
pip install "pyiceberg-dlf[rest-sigv4,pyarrow,pandas]"

説明

パッケージ名はインポート名と異なる：パッケージ名は pyiceberg-dlf ですが、インポート名は pyiceberg のままです。
pyiceberg-dlf と公式の pyiceberg は共存不可：両方のパッケージが import pyiceberg を提供します。一方をインストールすると、他方は警告なしに上書きされます。専用の仮想環境を使用するか、インストールする前に pip uninstall -y pyiceberg を実行してください。
rest-sigv4 は必須：この extra は boto3 とその他の署名関連の依存関係をインストールします。これがない場合、load_catalog() で ModuleNotFoundError: No module named 'boto3' が発生します。
pandas は任意：この例では scan.to_pandas() を使用しているため、DataFrame 出力が必要な場合は pandas extra を含めてください。
pyarrow は 22 未満である必要あり：pyiceberg-dlf[pyarrow] は pyarrow を 22 未満のバージョンに固定します。pyarrow 22 以降では、OSS へのアップロード時に aws-chunked encoding is not supported エラーが発生します。

コード例

次の Python スクリプトは、カタログに接続し、テーブルを作成し、データを書き込み、データを読み取り、その後テーブルを削除する方法を説明します。

スクリプト内のプレースホルダーを、ご自身の認証情報と設定に置き換えてください。

${regionId}：DLF サービスのリージョン (例：cn-hangzhou)。詳細については、「Iceberg REST サービスエンドポイント」をご参照ください。
${catalogName}：DLF カタログの名前。
${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 = 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": "${accessKeySecret}",
        "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-dlf         0.10.0.dev0
(myenv) root@iZbp1h4zr65vjcvz9w3080Z:~/workspace#