如何用AWS Lambda触发带特定过滤条件的Docker化ETL且不修改代码

核心结论 #

必须对ETL代码做少量修改，但改动极小，完全可以实现单一镜像复用，同时支持Fargate全量运行和Lambda带过滤条件的增量运行。

实现步骤 #

1. 理解Lambda运行Docker镜像的事件传递机制

Lambda运行自定义Docker镜像时，会通过两种方式传递触发事件：

• 将事件写入容器内的${LAMBDA_TASK_ROOT}/event.json文件（LAMBDA_TASK_ROOT是Lambda内置环境变量，默认值为/var/task）
• 通过标准输入（stdin）将事件传入容器

同时，Lambda会自动注入专属环境变量（如AWS_LAMBDA_FUNCTION_NAME），可用来判断当前运行环境。

2. 修改ETL入口逻辑，适配双环境

在ETL的启动脚本中增加环境判断分支：

• 检测到Lambda专属环境变量时，读取事件负载提取过滤条件，执行增量更新
• 未检测到则按原有逻辑执行全量ETL

以Python为例，修改后的入口脚本示例：

import os
import json

def main():
    # 判断是否处于Lambda运行环境
    if "AWS_LAMBDA_FUNCTION_NAME" in os.environ:
        # 方式1：读取Lambda生成的event.json文件
        event_path = os.path.join(os.environ["LAMBDA_TASK_ROOT"], "event.json")
        with open(event_path, "r") as f:
            event = json.load(f)
        
        # 方式2：从标准输入读取事件（二选一即可）
        # import sys
        # event = json.load(sys.stdin)
        
        # 提取触发时传入的过滤条件
        target_activity = event.get("target_activity_id")
        run_etl(filter_id=target_activity)
    else:
        # Fargate环境，执行全量ETL
        run_etl(filter_id=None)

def run_etl(filter_id=None):
    # 原有ETL核心逻辑，新增过滤分支
    if filter_id:
        print(f"执行增量更新：仅处理活动ID {filter_id}")
        # 写入按ID过滤的ETL逻辑
    else:
        print("执行全量ETL流程")
        # 原有全量处理逻辑

if __name__ == "__main__":
    main()

3. 调整Dockerfile，兼容Lambda Runtime

确保Dockerfile的CMD/ENTRYPOINT指向修改后的入口脚本，同时建议使用Lambda官方基础镜像（如Python、Node.js等），自动兼容Lambda Runtime API，无需额外配置：

# 以Python为例，使用Lambda官方基础镜像
FROM public.ecr.aws/lambda/python:3.11

# 复制ETL代码到Lambda指定目录
COPY . ${LAMBDA_TASK_ROOT}

# 安装依赖
RUN pip install -r requirements.txt

# 指定Lambda执行的入口函数（格式：脚本名.函数名）
CMD ["etl_entry.main"]

这个镜像既可以在Fargate通过docker-compose up正常启动，也能直接上传到ECR供Lambda使用。

4. 触发Lambda时传递过滤条件

通过API Gateway、CloudWatch Events、S3事件等触发Lambda时，传入包含过滤条件的JSON负载即可，示例：

{
  "target_activity_id": "ACT-20240501"
}

ETL代码会自动读取该参数，执行指定活动的更新。

补充说明 #

• 关于环境变量：Lambda支持配置静态环境变量，但仅适合存储固定参数（如数据库连接串），动态过滤条件建议通过事件负载传递，更灵活。
• 镜像复用性：修改后的代码完全兼容原有Fargate运行逻辑，无需维护两个镜像，彻底降低维护成本。

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