使用Glue Catalog创建Iceberg表的Schema构建问题求助

一、解决pydantic_core._pydantic_core.ValidationError错误 #

• 错误根源：table_input中的Schema字段使用了pyiceberg的Schema实例，但GlueCatalog.create_table的properties参数仅接受字典类型，非pyiceberg对象，触发了Pydantic类型验证失败。
• 修正操作：删除table_input中的Schema字段，pyiceberg会通过create_table的schema参数自动同步表结构到Glue Catalog，无需手动重复定义。

二、解决表创建后无列的问题 #

• 错误原因：
  1. 将完整的Glue Table配置（table_input）传入create_table的properties参数，导致pyiceberg无法正确解析并同步表结构。
  2. 重复创建GlueCatalog实例，第二个实例未指定warehouse参数，引发配置冲突。
• 修正操作：
  • 简化properties参数，仅传入Iceberg所需的额外属性（如存储格式相关配置），而非完整Glue Table配置。
  • 统一GlueCatalog实例的初始化逻辑，保证配置一致性。

三、完整修正代码 #

from pyiceberg.catalog import GlueCatalog
from pyiceberg.schema import Schema, NestedField
from pyiceberg.types import StringType, IntegerType
import boto3
from typing import Optional

def get_s3_path(bucket_name: str, database_name: Optional[str] = None, table_name: Optional[str] = None) -> str:
    result_path = f"s3://{bucket_name}"
    if database_name is not None:
        result_path += f"/{database_name}.db"
    if table_name is not None:
        result_path += f"/{table_name}"
    return result_path

def get_bucket_name() -> str:
    return bucket_name

# 统一初始化GlueCatalog实例
test_catalog = GlueCatalog(
    catalog_name,
    warehouse=get_s3_path(get_bucket_name()),
    **{"py-io-impl": "pyiceberg.io.fsspec.FsspecFileIO"}
)

table_schema_nested = Schema(
    NestedField(field_id=1, name="column1", field_type=StringType()),
    NestedField(field_id=2, name="column2", field_type=IntegerType(), required=False),
)

# 仅保留Iceberg需要的额外属性，而非完整Glue Table配置
table_properties = {
    "table_type": "EXTERNAL_TABLE",
    "input_format": "org.apache.hadoop.hive.ql.io.parquet.ParquetInputFormat",
    "output_format": "org.apache.hadoop.hive.ql.io.parquet.ParquetOutputFormat",
    "serialization_lib": "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"
}

def test_create_table(table_schema_nested: Schema, table_name, database_name):
    identifier = (database_name, table_name)
    # 先检查命名空间是否存在，避免删除不存在的命名空间报错
    try:
        test_catalog.drop_namespace(database_name)
    except Exception:
        pass
    # 创建命名空间
    test_catalog.create_namespace(
        namespace=database_name, 
        properties={"location": f"s3://{bucket_name}/{database_name}"}
    )
    # 创建Iceberg表
    table = test_catalog.create_table(
        identifier=identifier,
        schema=table_schema_nested,
        location=f"s3://{bucket_name}/{database_name}/{table_name}",
        properties=table_properties,
        partition_spec=["column1"]  # 直接通过pyiceberg指定分区
    )
    return table

# 初始化Glue客户端（仅用于验证表结构，无需用它操作分区）
glue_client = boto3.client(
    'glue',
    region_name=region,
    aws_access_key_id=aws_access_key_id,
    aws_secret_access_key=aws_secret_access_key
)

# 创建表
iceberg_table = test_create_table(table_schema_nested, table_name, database_name)

# 验证表结构
response = glue_client.get_tables(DatabaseName=database_name)
print("表结构信息:", response)

# 使用pyiceberg API写入数据（而非Glue的batch_create_partition，符合Iceberg规范）
data = [
    {'column1': 'Anurag', 'column2': 25},
]
iceberg_table.append(data)

额外说明 #

• 不要使用Glue的batch_create_partition操作Iceberg表：该接口是为Hive风格分区设计的，Iceberg使用自己的分区管理机制，直接用pyiceberg的append方法写入数据即可自动维护分区。
• 确保pyiceberg版本与AWS Glue兼容：建议使用pyiceberg 0.5.0及以上版本，避免API不匹配问题。

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