Glue PySpark固定宽度文件读取与校验问题咨询

1. 高效检查每行长度并过滤不符记录 #

• 借助Spark分布式计算能力，先将文件读取为单列数据（确保每行完整作为一个字段），再基于预设的总行长度（所有列长度之和）做过滤。
• 可以直接用Glue Dynamic Frame的filter方法，或转换为Spark DataFrame后执行filter操作，两种方式都能高效处理大数据量的行长度校验。

2. 读取固定宽度文件的最佳方式 #

• 优先用内置固定宽度解析器：通过create_dynamic_frame_from_options指定format="fixedwidth"，配合format_options配置列宽度，直接解析为多列Dynamic Frame，读取阶段即可完成基础格式校验。示例代码：

dyf = glueContext.create_dynamic_frame_from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://your-bucket/target-path/"]},
    format="fixedwidth",
    format_options={
        "fixedWidth": {"widths": [10, 15, 8], "trim": True}  # 替换为实际列长度
    }
)

• 先读为单列再处理：如果需要先过滤无效行再解析列，可按CSV格式读取，设置一个文件中不存在的分隔符（如"\u0000"），将每行完整存入col0字段，后续再做长度校验和列拆分。

3. 分离错误记录与有效记录的实现 #

你当前的代码已经能筛选出长度错误的记录，只需补充反向过滤得到有效记录即可，同时确保读取配置正确避免字段异常：

步骤1：正确读取文件为单列Dynamic Frame #

dynamicFramerawtxt = glueContext.create_dynamic_frame_from_options(
    connection_type="s3",
    connection_options={"paths": ["s3://your-bucket/txt-files/"]},
    format="csv",
    format_options={"separator": "\u0000", "multiline": False}
)

步骤2：计算总行长度并分离错误/有效记录 #

# 替换为实际各列长度之和
total_row_len = 10 + 15 + 8

# 创建错误记录的Dynamic Frame（ERROR_Dynamic帧）
error_dyf = dynamicFramerawtxt.filter(lambda x: len(x['col0']) != total_row_len)

# 保留长度正确的记录用于后续处理
valid_dyf = dynamicFramerawtxt.filter(lambda x: len(x['col0']) == total_row_len)

注：如果出现x['col0']取值异常，检查读取时的separator配置，确保没有将行内容拆分为多个字段，保证col0存储的是完整行文本。

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