如何用PyIceberg结合PyArrow在Python中创建分区表

现有如下DataFrame数据：

希望基于该DataFrame通过Python代码创建分区表，目前已有可成功创建非分区表的代码：

import pandas as pd
import pyiceberg
from pyiceberg.catalog import load_catalog
import pyarrow as pa

catalog = load_catalog("glue_catalog",type='glue')
df = pd.read_csv(r'Test.csv')

df['business_time']=pd.to_datetime(df['business_time'], utc=True).dt.floor('us').astype(pd.ArrowDtype(pa.timestamp("us", tz='UTC')))

pat = pa.Table.from_pandas(df)
t = catalog.create_table_if_not_exists('test_schema.test_partition', pat.schema)                                    
t.append(pat)

想要实现如下SQL语句等效的分区表创建效果：

CREATE TABLE glue_catalog.test_schema.test_partition (
   business_time timestamp(6) with time zone,
   value double
)
WITH (
   format = 'PARQUET',
   format_version = 2,
   partitioning = ARRAY['day(business_time)']
);

尝试添加参数partition_spec=pyiceberg.partitioning.DayTransform('business_time')时出现验证错误，求简单的字段指定方式实现按business_time字段按天分区。

需要将DayTransform包装到PartitionSpec对象中，而非直接传入单个Transform。修改后的代码如下：

import pandas as pd
import pyiceberg
from pyiceberg.catalog import load_catalog
import pyarrow as pa
from pyiceberg.partitioning import PartitionSpec, DayTransform

catalog = load_catalog("glue_catalog", type='glue')
df = pd.read_csv(r'Test.csv')

# 处理时间字段类型，确保与SQL定义匹配
df['business_time'] = pd.to_datetime(df['business_time'], utc=True).dt.floor('us').astype(pd.ArrowDtype(pa.timestamp("us", tz='UTC')))

pat = pa.Table.from_pandas(df)

# 创建按business_time日期分区的规则
partition_spec = PartitionSpec(DayTransform("business_time"))

# 创建分区表，指定分区规则和存储格式配置
t = catalog.create_table_if_not_exists(
    'test_schema.test_partition',
    schema=pat.schema,
    partition_spec=partition_spec,
    properties={
        "format": "PARQUET",
        "format_version": "2"
    }
)                                    
t.append(pat)

关键说明 #

• PartitionSpec是包装分区转换规则的容器，单个时间转换规则必须放入该容器才能被正确识别。
• 通过properties参数指定存储格式和版本，与目标SQL中的WITH配置完全对应。
• 你之前对business_time字段的类型处理已经符合要求，确保了和SQL定义的timestamp(6) with time zone匹配。

内容的提问来源于stack exchange，提问作者user23873134