如何在AWS Lambda中无集群运行Spark任务并实现S3触发处理

一、前置准备 #

• 确认Lambda运行时：推荐使用Python 3.9/3.10（与主流PySpark版本兼容）
• 构建PySpark Lambda层：解决Spark依赖问题，避免打包过大的Lambda部署包
• 配置IAM角色：给Lambda分配S3读写、CloudWatch日志权限

二、构建PySpark Lambda层 #

1. 本地创建目录结构：python/lib/pythonX.X/site-packages（X.X对应Python版本，如3.9）
2. 安装PySpark及依赖：

   pip install pyspark==3.3.0 py4j==0.10.9.5 -t python/lib/python3.9/site-packages
   

3. 打包成zip：

   cd python && zip -r pyspark-layer.zip .
   

4. 上传到Lambda层：在Lambda控制台的「层」页面创建新层，上传zip包，指定对应Python运行时

三、配置Lambda函数 #

1. 创建新Lambda函数，选择对应Python运行时
2. 关联PySpark层：在函数的「层」选项卡添加刚才创建的PySpark层
3. 调整资源配置：
   • 内存：至少1GB（Spark运行需要足够内存，可根据数据量上调至2-4GB）
   • 超时：设置为5-15分钟（根据任务复杂度调整）
   • 临时存储：默认512MB，若处理大文件可上调至10GB

四、编写Spark处理代码 #

以下是Python示例代码，实现读取S3上传的CSV、处理并写入输出桶：

import pyspark.sql.functions as F
from pyspark.sql import SparkSession

def lambda_handler(event, context):
    # 初始化本地模式SparkSession
    spark = SparkSession.builder \
        .master("local[*]") \
        .appName("S3SparkProcessor") \
        .config("spark.local.dir", "/tmp")  # 指定Lambda临时目录存储Spark中间文件
        .getOrCreate()
    
    # 解析S3触发事件，获取输入文件路径
    s3_event = event['Records'][0]['s3']
    input_bucket = s3_event['bucket']['name']
    input_key = s3_event['object']['key']
    input_path = f"s3://{input_bucket}/{input_key}"
    
    # 读取CSV文件（根据实际情况调整header、inferSchema参数）
    raw_df = spark.read.csv(input_path, header=True, inferSchema=True)
    
    # ------------------- 数据处理逻辑（示例）-------------------
    # 示例：添加处理时间戳、筛选符合条件的数据
    processed_df = raw_df \
        .withColumn("processed_at", F.current_timestamp()) \
        .filter(F.col("age") > 18)  # 假设CSV有age列，筛选成年数据
    # ---------------------------------------------------------
    
    # 写入输出S3桶（可替换为parquet等更高效格式）
    output_bucket = "your-output-bucket-name"
    output_key_prefix = f"processed/{input_key.split('/')[-1].replace('.csv', '')}"
    output_path = f"s3://{output_bucket}/{output_key_prefix}"
    
    processed_df.write.mode("overwrite").csv(output_path, header=True)
    
    # 关闭SparkSession释放资源
    spark.stop()
    
    return {
        "statusCode": 200,
        "body": f"Successfully processed {input_key}. Output saved to {output_path}"
    }

五、配置S3触发事件 #

1. 进入输入CSV的S3桶控制台，打开「属性」→「事件通知」
2. 创建新通知：
   • 名称：自定义（如csv-upload-trigger）
   • 事件类型：勾选「所有对象创建事件」（或仅选「Put」）
   • 前缀/后缀：若仅触发file.csv上传，可设置后缀为.csv，或直接指定对象键为file.csv
   • 目标：选择「Lambda函数」，并指定刚才创建的Spark处理函数
3. 保存配置，完成触发链路搭建

六、常见问题与优化 #

• 内存不足：上调Lambda内存配置，同时在SparkSession中调整spark.driver.memory参数（如.config("spark.driver.memory", "1g")）
• 超时：延长Lambda超时时间，优化数据处理逻辑（如减少不必要的转换）
• 依赖缺失：检查Lambda层的Python版本与函数运行时是否匹配，确认PySpark包已正确打包
• 数据规模限制：Lambda临时存储最大10GB，若处理超大型文件，建议改用EMR Serverless或Glue ETL

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