使用AWS Glue批量迁移S3表到Redshift的类型及Schema问题求助

1. 修复Redshift Schema映射错误 #

你遇到的schema1 is not defined报错以及Schema不生效的问题，是两处参数配置错误导致的：

• connection_options里的database参数只能填Redshift的数据库名，不能携带Schema信息，Schema需要拼接到dbtable参数中
• 拼接schema1.tableName时要把schema1作为字符串处理，正确的f-string拼接写法是f"schema1.{tableName}"，直接写schema1.tableName会被Python识别为schema1变量的属性调用，自然会抛出变量未定义的错误。

2. 实现按表条件执行类型转换 #

只需要预先定义一个转换规则配置字典，将需要做类型转换的表名作为key，对应表的resolveChoice所需的specs规则作为value，循环时判断当前表是否在配置字典中，存在则先执行类型转换再写入即可。

修正后完整代码 #

import boto3

# 预先定义类型转换规则：key为需要转换的表名，value为对应列的转换规则
type_convert_config = {
    "table1": [
        ("column1", "cast:char"),
        ("column2", "cast:varchar"),
        ("column3", "cast:varchar"),
    ],
    "table2": [
        ("xxx_col", "cast:int"),
        ("yyy_col", "cast:timestamp"),
    ]
    # 其他需要转换的表按同样格式添加即可
}

client = boto3.client("glue", region_name="us-east-1")
databaseName = "db1_g"
Tables = client.get_tables(DatabaseName=databaseName)
tableList = Tables["TableList"]

redshift_db_name = "db1"
redshift_schema = "schema1"

for table in tableList:
    tableName = table["Name"]
    # 读取表数据
    datasource = glueContext.create_dynamic_frame.from_catalog(
        database="db1_g", 
        table_name=tableName, 
        transformation_ctx=f"datasource_{tableName}"
    )
    
    # 判断当前表是否需要做类型转换
    if tableName in type_convert_config:
        datasource = datasource.resolveChoice(specs=type_convert_config[tableName])
    
    # 写入Redshift
    datasink = glueContext.write_dynamic_frame.from_jdbc_conf(
        frame=datasource,
        catalog_connection="redshift",
        connection_options={
            "dbtable": f"{redshift_schema}.{tableName}",
            "database": redshift_db_name,
        },
        redshift_tmp_dir=args["TempDir"],
        transformation_ctx=f"datasink_{tableName}",
    )

job.commit()

补充说明 #

• transformation_ctx按表名做唯一标识，可避免Glue任务快照恢复时出现冲突
• 后续新增需要转换的表，仅需在type_convert_config里添加对应配置即可，无需修改循环逻辑
• 如果有表需要写入不同的Schema，可把配置字典的value改成包含Schema、转换规则的结构体，即可灵活适配多Schema场景

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