Paimon Python SDK (pypaimon) 詳細使用指南

作者：隨川 2025-07-24 07:10:31

Pypaimon 作為 Apache Paimon 的 Python 客戶端，填補了 Python 生態與數據湖之間的鴻溝，使數據科學家和 Python 開發者能夠直接操作流批一體數據湖。

一、pypaimon 簡介

pypaimon 是 Apache Paimon 數據湖的 Python 客戶端 SDK，基于 Py4J 實現 Python 與 Java 代碼的橋接，允許開發者通過 Python API 操作 Paimon 數據湖。作為 Apache Paimon 的重要組件，pypaimon 繼承了 Paimon 的核心特性：流批一體存儲、實時數據更新、ACID 事務支持和低成本存儲，同時提供 Python 生態友好的接口，支持與 PyArrow、Pandas 等數據科學工具無縫集成。

核心定位：

技術橋梁：連接 Python 生態與 Paimon 數據湖，支持 Python 開發者直接操作數據湖表
輕量化集成：無需編寫 Java 代碼即可利用 Paimon 的 LSM 樹存儲結構和高效 compaction 機制
多場景適配：適用于實時數據入湖、批式數據處理、OLAP 查詢加速等場景

二、環境準備與安裝

1. 系統要求

依賴項	版本要求
Python	3.8 及以上
JRE	1.8
Hadoop 環境	可選（本地測試可省略）
PyArrow	推薦 7.0+
Pandas	推薦 1.3+

2. 安裝方式

(1) 阿里云 DLF 專用版本（推薦生產環境）

# 下載 pypaimon_dlf2 安裝包
wget https://help.aliyun.com/zh/dlf/dlf-2-0/use-cases/pypaimon-dlf-for-data-into-the-lake
pip3 install pypaimon_dlf2-0.3.dev0.tar.gz

(2) 官方開發版（適合測試）

pip install paimon-python==0.9.0.dev1

2. 環境驗證

# 檢查 Java 環境
import os
assert'JAVA_HOME'in os.environ,"請配置 JAVA_HOME 環境變量"

# 驗證安裝
from pypaimon import Schema
print("pypaimon 安裝成功")

三、核心 API 與基礎操作

1. Catalog 管理

Catalog 是 Paimon 數據湖的元數據入口，用于管理數據庫和表。pypaimon 支持多種 Catalog 類型，包括本地文件系統、HDFS、阿里云 DLF 等。

創建 DLF Catalog（阿里云場景）：

from pypaimon.py4j import Catalog

catalog_options ={
'metastore':'dlf-paimon',
'dlf.region':'cn-hangzhou',
'dlf.endpoint':'dlf.cn-hangzhou.aliyuncs.com',
'dlf.catalog.id':'your-catalog-id',
'dlf.catalog.accessKeyId':'your-ak',
'dlf.catalog.accessKeySecret':'your-sk',
'max-workers':'4'# 并行讀取線程數
}

catalog = Catalog.create(catalog_options)

2. 數據庫與表操作

(1) 創建數據庫

# 創建數據庫（忽略已存在錯誤）
catalog.create_database(
    name='paimon_demo',
    ignore_if_exists=True
)

(2) 定義表 Schema

通過 PyArrow 定義表結構，支持分區鍵、主鍵和表屬性配置：

import pyarrow as pa
from pypaimon import Schema

# 定義 PyArrow Schema
pa_schema = pa.schema([
('dt', pa.string()),
('user_id', pa.int64()),
('order_id', pa.int64()),
('amount', pa.float64())
])

# 轉換為 Paimon Schema
table_schema = Schema(
    pa_schema=pa_schema,
    partition_keys=['dt'],# 分區鍵
    primary_keys=['dt','order_id'],# 主鍵
    options={
'bucket':'8',# 分桶數
'file.format':'parquet'# 文件格式
},
    comment='電商訂單事實表'
)

(3) 創建表

# 在指定數據庫創建表
catalog.create_table(
    identifier='paimon_demo.orders',
    schema=table_schema,
    ignore_if_exists=True
)

# 獲取表對象
table = catalog.get_table('paimon_demo.orders')

3. 數據寫入與提交

pypaimon 支持 PyArrow Table 和 Pandas DataFrame 兩種寫入格式，通過兩階段提交保證數據一致性：

import pandas as pd

# 準備測試數據
data ={
'dt':['2024-01-01','2024-01-01','2024-01-02'],
'user_id':[1001,1002,1001],
'order_id':[10001,10002,10003],
'amount':[299.5,159.0,499.9]
}
df = pd.DataFrame(data)

# 創建寫入器
write_builder = table.new_batch_write_builder()
table_write = write_builder.new_write()
table_commit = write_builder.new_commit()

try:
# 寫入 Pandas DataFrame
    table_write.write_pandas(df)
# 準備提交
    commit_msg = table_write.prepare_commit()
# 執行提交
    table_commit.commit(commit_msg)
finally:
# 釋放資源
    table_write.close()
    table_commit.close()

4. 數據查詢與過濾

支持謂詞下推和投影優化，通過 ReadBuilder 配置查詢參數：

# 創建讀取器
read_builder = table.new_read_builder()

# 構建過濾條件 (dt = '2024-01-01')
predicate_builder = read_builder.new_predicate_builder()
predicate = predicate_builder.equal('dt','2024-01-01')
read_builder = read_builder.with_filter(predicate)

# 執行查詢
table_scan = read_builder.new_scan()
splits = table_scan.plan().splits()# 獲取數據分片

# 轉換為 PyArrow Table
table_read = read_builder.new_read()
pa_table = table_read.to_arrow(splits)

# 轉換為 Pandas DataFrame
result_df = pa_table.to_pandas()
print(result_df)

四、高級特性與性能優化

1. 并發控制與事務

pypaimon 采用樂觀并發控制，通過兩階段提交協議保證寫入原子性。對于對象存儲（如 S3/OSS），需額外配置：

# 啟用元數據鎖（對象存儲必需）
catalog_options['lock.enabled']='true'
catalog_options['metastore']='jdbc'# 使用 JDBC 元存儲

2. 數據類型映射

pypaimon 通過 PyArrow 實現 Python 與 Paimon 數據類型的自動映射：

Python 類型	PyArrow 類型	Paimon 類型
int	pa.int64()	BIGINT
float	pa.float64()	DOUBLE
str	pa.string()	STRING
datetime.datetime	pa.timestamp('ns')	TIMESTAMP
list	pa.list_(pa.int64())	ARRAY<BIGINT>

3. 性能調優參數

參數	說明	推薦值
max-workers	并行讀取線程數	4-8
bucket	分桶數（主鍵表）	8-32
compaction.delta-commits	增量壓縮觸發閾值	10
file.index.bloom-filter	啟用布隆過濾器索引	'user_id'

五、典型應用場景

1. 實時數據入湖（CDC 同步）

通過 Debezium 捕獲 MySQL 變更數據，經 Flink 處理后寫入 Paimon，pypaimon 負責批式補數據：

# 補傳歷史數據
historical_df = pd.read_csv('historical_orders.csv')
table_write.write_pandas(historical_df)
table_commit.commit(table_write.prepare_commit())

2. 流批一體分析

同一份數據同時支持批式報表和實時查詢：

# 批式查詢（T+1報表）
batch_read = table.new_read_builder().with_snapshot('20240101').build()

# 實時查詢（實時dashboard）
stream_read = table.new_read_builder().with_start_snapshot('LATEST').build()

3. 機器學習樣本存儲

存儲特征數據并支持高效讀取：

# 讀取特征數據用于模型訓練
features = table_read.to_arrow(splits).to_pandas()
X = features[['user_age','order_count']]
y = features['label']

六、注意事項與優秀實踐

1. 環境依賴

確保 JRE 8 環境變量配置正確：export JAVA_HOME=/path/to/jre8
本地測試推薦使用 Flink 預綁定的 Hadoop jar：export HADOOP_CLASSPATH=$(flink classpath)

2. 數據一致性

寫入后必須調用 commit() 方法，否則數據不會持久化
多writer場景需避免同一主鍵并發寫入，可能導致快照沖突

3. 資源配置

大表查詢建議設置 max-workers=8 提升并行度
內存受限場景啟用 spill 機制：sort-spill-threshold=10

pypaimon 作為 Apache Paimon 的 Python 客戶端，填補了 Python 生態與數據湖之間的鴻溝，使數據科學家和 Python 開發者能夠直接操作流批一體數據湖。其核心優勢在于：

簡單易用：Python 友好的 API 設計，降低數據湖使用門檻
生態融合：無縫對接 Pandas、PyArrow 等數據科學工具
性能卓越：繼承 Paimon 的 LSM 樹結構和高效 compaction 機制

責任編輯：趙寧寧來源：大數據技能圈