Data Lake Formation:PyPaimon による DLF Paimon テーブルに対するスタンドアロン SQL クエリの実行

前提条件

CATALOG_OPTIONS = {
    "metastore": "rest",
    "uri": "http://<DLF-ENDPOINT>",
    "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>",
}

DataFusion への Paimon カタログの登録

PaimonCatalog は DataFusion カタログプロバイダープロトコルを実装しています。SessionContext に登録すると、3部構成の名前 paimon.<database>.<table> を使用して Paimon テーブルを参照できます。

from datafusion import SessionContext
from pypaimon_rust.datafusion import PaimonCatalog

ctx = SessionContext()
paimon = PaimonCatalog(CATALOG_OPTIONS)
ctx.register_catalog_provider("paimon", paimon)

print("catalogs:", ctx.catalog_names())
# -> {'paimon', 'datafusion'}

# paimon カタログ内のすべてのデータベース (スキーマ) を一覧表示
print("databases:", ctx.catalog("paimon").names())

# default データベース内のすべてのテーブルを一覧表示
print("tables in default:", ctx.catalog("paimon").schema("default").names())

Paimon テーブルへのクエリ

以下の例では、employees テーブル (列 id、name、dept、salary) と departments テーブルがすでに存在することを前提としています。テーブル作成とデータ取り込みのコードについては、このドキュメントの最後にある [完全な例](#complete-example) をご参照ください。

クエリ結果は、.to_pylist()、.to_arrow_table()、.to_pandas() などのメソッドを使用して、さまざまなデータ形式に変換できます。

rows = ctx.sql("SELECT * FROM paimon.default.employees ORDER BY id").to_pylist()
print(rows[0])
# -> {'id': 1, 'name': 'Alice', 'dept': 'eng', 'salary': 9000}

eng = ctx.sql("""
    SELECT id, name, salary
    FROM paimon.default.employees
    WHERE dept = 'eng'
    ORDER BY salary DESC
""").to_pylist()
print(eng)
# -> [{'id': 4, 'name': 'Dan', 'salary': 9500},
#     {'id': 1, 'name': 'Alice', 'salary': 9000},
#     {'id': 2, 'name': 'Bob', 'salary': 8500}]

agg = ctx.sql("""
    SELECT dept, COUNT(*) AS n, AVG(salary) AS avg_sal
    FROM paimon.default.employees
    GROUP BY dept
    ORDER BY dept
""").to_pylist()
print(agg)
# -> [{'dept': 'eng',   'n': 3, 'avg_sal': 9000.0},
#     {'dept': 'ops',   'n': 1, 'avg_sal': 6500.0},
#     {'dept': 'sales', 'n': 2, 'avg_sal': 7100.0}]

result = ctx.sql("""
    SELECT a.id AS id_a, b.id AS id_b, a.name
    FROM paimon.default.employees a
    JOIN paimon.default.employees b
      ON a.id = b.id
    WHERE a.dept = 'eng'
    ORDER BY a.id
""").to_pylist()

result = ctx.sql("""
    SELECT e.id, e.name, e.dept, d.dept_full
    FROM paimon.default.employees e
    JOIN paimon.default.departments d
      ON e.dept = d.dept
    WHERE e.dept = 'eng'
    ORDER BY e.id
""").to_pylist()

クエリ結果の変換

クエリ結果を複数の形式に変換して、さまざまな下流の処理フレームワークと統合できます。

arrow_tbl = ctx.sql(
    "SELECT id, salary FROM paimon.default.employees ORDER BY id"
).to_arrow_table()
print(arrow_tbl.num_rows, "rows")
print(arrow_tbl.schema)

df = ctx.sql(
    "SELECT * FROM paimon.default.employees ORDER BY id"
).to_pandas()
print(df.head())

import daft

daft_df = daft.from_arrow(arrow_tbl)
daft_df.show()

完全な例

次の例は、PyPaimon を使用してテーブルを作成し、データを書き込み、次に DataFusion にカタログを登録して SQL クエリを実行するという、エンドツーエンドのワークフローを示しています。

from __future__ import annotations

import os
from datetime import datetime

import pyarrow as pa
from datafusion import SessionContext
from pypaimon_rust.datafusion import PaimonCatalog
from pypaimon import CatalogFactory, Schema


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:
    # 1) PyPaimon を使用してテーブルを作成し、データを書き込む
    pcat = CatalogFactory.create(CATALOG_OPTIONS)
    pcat.create_database(DATABASE, True)

    table_name = f"sql_demo_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    full_name = f"{DATABASE}.{table_name}"
    pa_schema = pa.schema([
        pa.field("id",     pa.int32(), nullable=False),
        pa.field("name",   pa.string()),
        pa.field("dept",   pa.string()),
        pa.field("salary", pa.int64()),
    ])
    schema = Schema.from_pyarrow_schema(
        pa_schema,
        options={"bucket": "-1", "file.format": "parquet"},
    )
    pcat.create_table(full_name, schema, True)

    tbl = pcat.get_table(full_name)
    wb = tbl.new_batch_write_builder()
    w = wb.new_write()
    w.write_arrow(pa.Table.from_pydict({
        "id":     [1, 2, 3, 4, 5, 6],
        "name":   ["Alice", "Bob", "Carol", "Dan", "Eve", "Frank"],
        "dept":   ["eng", "eng", "sales", "eng", "sales", "ops"],
        "salary": [9000, 8500, 7000, 9500, 7200, 6500],
    }, schema=pa_schema))
    wb.new_commit().commit(w.prepare_commit())
    w.close()
    print(f"created + wrote 6 rows to {full_name}")

    # 2) DataFusion にカタログを登録する
    ctx = SessionContext()
    ctx.register_catalog_provider("paimon", PaimonCatalog(CATALOG_OPTIONS))

    # 3) SQL クエリを実行する
    rows = ctx.sql(
        f"SELECT * FROM paimon.{DATABASE}.{table_name} ORDER BY id"
    ).to_pylist()
    print("all rows:", rows)

    eng = ctx.sql(
        f"SELECT id, name, salary FROM paimon.{DATABASE}.{table_name} "
        f"WHERE dept = 'eng' ORDER BY salary DESC"
    ).to_pylist()
    print("eng:", eng)

    agg = ctx.sql(
        f"SELECT dept, COUNT(*) AS n, AVG(salary) AS avg_sal "
        f"FROM paimon.{DATABASE}.{table_name} GROUP BY dept ORDER BY dept"
    ).to_pylist()
    print("by dept:", agg)


if __name__ == "__main__":
    main()

PyPaimon の SQL CLI の使用

PyPaimon には、組み込みの SQL CLI が含まれています。YAML 設定ファイルでカタログ接続を指定した後、対話型 REPL を開始するか、単一の SQL ステートメントを実行できます。

カタログ設定を catalog.yaml として保存します：

metastore: rest
uri: http://<DLF-VPC-ENDPOINT>
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>

paimon コマンドを使用してクエリを実行します：

# 対話型 REPL を開始
paimon --config /path/to/catalog.yaml sql

# 単一の SQL ステートメントを実行
paimon --config /path/to/catalog.yaml sql "SELECT * FROM default.employees LIMIT 5"

# 出力形式を指定 (デフォルト： table、json もサポート)
paimon --config /path/to/catalog.yaml sql --format json "SELECT * FROM default.employees"

出力形式や結果のエクスポートなどの追加オプションについては、paimon sql --help を実行すると、すべてのパラメーターリストが表示されます。

SQL クエリインターフェイスの比較

PyPaimon は、使用パターンが異なる 2 つの SQL クエリインターフェイスを提供しています。このドキュメントでは、PaimonCatalog + DataFusion 方式について説明します。SQLContext 方式については、PyPaimon API ドキュメントをご参照ください。