# 下載 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

pip install paimon-python==0.9.0.dev1

# 檢查 Java 環(huán)境
import os
assert'JAVA_HOME'in os.environ,"請(qǐng)配置 JAVA_HOME 環(huán)境變量"

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

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'# 并行讀取線程數(shù)
}

catalog = Catalog.create(catalog_options)

# 創(chuàng)建數(shù)據(jù)庫(kù)（忽略已存在錯(cuò)誤）
catalog.create_database(
    name='paimon_demo',
    ignore_if_exists=True
)

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())
])

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

# 在指定數(shù)據(jù)庫(kù)創(chuàng)建表
catalog.create_table(
    identifier='paimon_demo.orders',
    schema=table_schema,
    ignore_if_exists=True
)

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

import pandas as pd

# 準(zhǔn)備測(cè)試數(shù)據(jù)
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)

# 創(chuàng)建寫入器
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)
# 準(zhǔn)備提交
    commit_msg = table_write.prepare_commit()
# 執(zhí)行提交
    table_commit.commit(commit_msg)
finally:
# 釋放資源
    table_write.close()
    table_commit.close()

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

# 構(gòu)建過濾條件 (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)

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

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

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

# 啟用元數(shù)據(jù)鎖（對(duì)象存儲(chǔ)必需）
catalog_options['lock.enabled']='true'
catalog_options['metastore']='jdbc'# 使用 JDBC 元存儲(chǔ)

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

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

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

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