数据湖构建：使用 PyPaimon 单机 SQL 查询 Paimon 表

前提条件

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>",
}

注册 Paimon Catalog 到 DataFusion

PaimonCatalog 实现了 DataFusion 的 catalog provider 协议。注册到 SessionContext 之后，即可通过 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 catalog 下所有 database（schema）
print("databases:", ctx.catalog("paimon").names())

# 列出 default database 下所有 table
print("tables in default:", ctx.catalog("paimon").schema("default").names())

查询 Paimon 表

以下示例假设已存在 employees 表（包含 id、name、dept、salary 四列）和 departments 表。建表与写入代码见完整示例。

查询结果可以通过 .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) 注册 Catalog 到 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，通过 YAML 配置文件指定 Catalog 连接后，即可启动交互式 REPL 或执行单条 SQL。

将 Catalog 配置保存为 catalog.yaml：

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>

通过 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 提供两种 SQL 查询入口，使用方式存在差异。本文介绍的是 PaimonCatalog + DataFusion 方式，SQLContext 方式请参见 PyPaimon API 文档。