如何在AWS Glue中运行多Py文件压缩包ETL任务及优化结构

一、作业基础设置（打包与上传） #

1. 正确打包py_files.zip
   • 压缩包内直接包含所有需要的文件/模块，不要嵌套额外根文件夹。比如解压后应看到：__init__.py、config.json、transformations.py，而非py_files/文件夹下的这些文件。
   • 空的__init__.py必须存在，这是Python识别模块的必要条件。
2. 上传至S3
   • 将py_files.zip上传到你的S3存储桶，示例路径：s3://your-glue-artifacts/py_modules/py_files.zip
3. Glue作业配置
   • 打开Glue作业编辑页面，在**"作业参数"**中添加键值对：
     • 键：--extra-py-files
     • 值：s3://your-glue-artifacts/py_modules/py_files.zip
   • 或在**"附加Python模块"**字段中直接填入上述S3路径（部分Glue版本支持可视化配置）。
   • 确认Glue作业的IAM角色拥有该S3路径的s3:GetObject权限。

二、脚本配置与导入问题解决 #

1. 正确导入模块 #

如果zip包直接包含transformations.py，在Glue主脚本中直接导入：

from transformations import transform_customer_data, transform_order_data

如果是子模块结构（比如transformations/clean.py），确保子文件夹下有__init__.py，导入方式：

from transformations.clean import remove_null_values

2. 读取config.json #

用Python的pkg_resources模块读取压缩包内的配置文件，避免路径问题：

import json
import pkg_resources

# 读取zip内的config.json
config_raw = pkg_resources.resource_string(__name__, "config.json").decode("utf-8")
config = json.loads(config_raw)

# 使用配置项
source_bucket = config["source_s3_bucket"]
target_table = config["target_glue_table"]

3. 导入失败排查 #

• 检查zip包结构：解压后确认文件层级正确，无多余根目录。
• 验证S3路径：确保作业参数中的路径与实际S3路径完全一致，包括大小写。
• 权限检查：确认Glue角色的IAM策略允许读取该S3路径。
• 检查导入语句：Python区分大小写，确保模块名与文件名完全匹配。

三、避免单长脚本的文件结构最佳实践 #

按功能职责拆分模块，推荐结构如下：

your-etl-module/
├── __init__.py
├── main.py          # Glue作业入口，调度各模块执行
├── config/
│   ├── __init__.py
│   ├── config.json  # 环境配置（数据源、目标地址等）
│   └── constants.py # 常量定义（比如日期格式、字段名）
├── transformations/
│   ├── __init__.py
│   ├── clean.py     # 数据清洗逻辑
│   ├── enrich.py    # 数据增强逻辑
│   └── aggregate.py # 数据聚合逻辑
├── core/
│   ├── __init__.py
│   ├── source.py    # 数据源读取（S3、JDBC等）
│   └── sink.py      # 数据输出（Glue表、S3等）
└── utils/
    ├── __init__.py
    ├── logger.py    # 自定义日志配置
    └── validator.py # 数据校验工具

结构优势： #

• 单一职责：每个文件/模块只负责一类功能，便于维护和调试。
• 复用性：工具类、核心读写逻辑可在多个ETL作业中复用。
• 可读性：新成员能快速通过目录结构理解代码分工。

打包注意事项： #

将整个your-etl-module文件夹内的所有内容打包成zip（不要包含外层your-etl-module文件夹），再按之前的方法配置到Glue作业中，主脚本直接调用main.py的入口函数即可。

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