Glue Job读取S3多文件失败，单文件可正常写入Redshift问题排查

以下是针对该问题的常见排查点和解决操作：

1. 强制指定数据Schema，避免自动推断冲突 #

多个文件可能存在字段类型不一致、字段缺失的情况，Glue自动推断Schema时会出现矛盾，导致任务中断。

• 读取S3数据时手动定义Schema：

from awsglue.schema import StructType, StructField
from awsglue.type import StringType, IntegerType, DoubleType

custom_schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("value", DoubleType(), True)
])

dynamic_frame = glueContext.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://your-bucket/folder-path/"]},
    format="csv",  # 替换为你的文件格式（parquet/json等）
    format_options={"withHeader": True},
    schema=custom_schema
)

• 若仍有类型冲突，使用resolveChoice处理：

resolved_frame = dynamic_frame.resolveChoice(choice="make_cols", transformation_ctx="resolve_choice")

2. 过滤S3文件夹中的无效/隐藏文件 #

文件夹中可能存在_SUCCESS、.DS_Store或临时文件，这些文件会干扰数据读取：

• 读取时添加文件过滤规则：

connection_options={
    "paths": ["s3://your-bucket/folder-path/"],
    "recurse": True,  # 如需递归读取子文件夹
    "excludePattern": "_SUCCESS|\\.DS_Store|.*\\.tmp"
}

3. 检查Redshift写入的临时目录权限 #

Glue写入Redshift时会先将数据转存到临时S3目录，若Redshift没有该目录的读写权限，批量写入会失败：

• 确保redshift_tmp_dir配置的S3路径已授权给Redshift集群的IAM角色：

glueContext.write_dynamic_frame.from_jdbc_conf(
    frame=resolved_frame,
    catalog_connection="your-redshift-connection",
    connection_options={
        "dbtable": "public.target_table",
        "redshiftTmpDir": "s3://your-bucket/tmp-redshift/",
        "maxBatchSize": "10000"  # 调整批量写入大小，适配数据量
    },
    transformation_ctx="write_to_redshift"
)

• 确认Redshift集群的IAM角色拥有该S3路径的s3:PutObject、s3:GetObject、s3:DeleteObject权限。

4. 处理数据中的异常值 #

多文件可能包含空值、特殊字符（如换行符、控制字符），触发Redshift的约束校验：

• 清理空值：

cleaned_frame = dynamic_frame.drop_nulls(["id"])  # 针对非空字段
# 或填充空值
filled_frame = dynamic_frame.fill_nulls({"value": 0.0})

• 移除特殊字符（以字符串字段为例）：

from pyspark.sql.functions import regexp_replace

spark_df = dynamic_frame.toDF()
cleaned_spark_df = spark_df.withColumn("name", regexp_replace("name", "[\\r\\n\\t]", ""))
cleaned_frame = DynamicFrame.fromDF(cleaned_spark_df, glueContext, "cleaned_frame")

5. 调整Glue Job的资源配置 #

多文件数据量较大时，默认Worker配置可能内存不足导致任务崩溃：

• 增加Worker数量（如从2个调整为5个）或升级Worker类型（如从Standard改为G.2X）；
• 在Job参数中添加--enable-metrics，查看CloudWatch日志中的内存、CPU使用率，确认是否存在资源瓶颈；
• 若出现OOM错误，调整Spark参数：--conf spark.driver.maxResultSize=4g、--conf spark.executor.memory=8g。

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