如何将Step Functions变量传入EMR Serverless PySpark的entryPointArguments

1. Step Functions状态机配置：正确引用动态变量 #

在Step Functions的StartJobRun任务参数中，使用States路径语法直接引用状态机中的变量，替换硬编码值。建议采用--参数名 参数值的格式传入，方便PySpark的argparse解析。

示例状态机JSON片段： #

{
  "States": {
    "RunEMRServerlessSparkJob": {
      "Type": "Task",
      "Resource": "arn:aws:states:::emr-serverless:startJobRun.sync",
      "Parameters": {
        "ApplicationId": "<你的EMR Serverless应用ID>",
        "ExecutionRoleArn": "<你的执行角色ARN>",
        "JobDriver": {
          "SparkSubmit": {
            "EntryPoint": "s3://<你的代码存储路径>/main.py",
            "EntryPointArguments": [
              "--today_date", "$.today_date",
              "--source", "$.source",
              "--tuned_parameters", "${States.JsonToString($.tuned_parameters)}"
            ]
          }
        }
      },
      "End": true
    }
  }
}

• 字符串类型变量（如today_date、source）直接用$.变量名引用；
• 复杂类型（如字典或JSON结构的tuned_parameters），用States.JsonToString()转为字符串，避免参数传递时的格式错误。

2. PySpark代码：用argparse接收并解析参数 #

在PySpark代码中，通过argparse定义对应参数，接收Step Functions传入的值，复杂类型参数再转回JSON对象。

示例PySpark代码片段： #

import argparse
import json
from pyspark.sql import SparkSession

def main():
    # 解析命令行参数
    parser = argparse.ArgumentParser()
    parser.add_argument("--today_date", required=True, help="日期参数")
    parser.add_argument("--source", required=True, help="数据源标识")
    parser.add_argument("--tuned_parameters", required=True, help="调优参数JSON字符串")
    args = parser.parse_args()

    # 初始化SparkSession
    spark = SparkSession.builder.appName("DynamicArgsDemo").getOrCreate()

    # 使用参数
    print(f"今日日期: {args.today_date}")
    print(f"数据源: {args.source}")
    
    # 解析复杂参数
    tuned_params = json.loads(args.tuned_parameters)
    print(f"调优参数: {tuned_params}")

    # 后续业务逻辑...

if __name__ == "__main__":
    main()

3. 常见问题排查 #

• 变量路径错误：检查Step Functions输入中的变量是否存在，路径是否正确（比如嵌套层级变量要写成$.parent.child）；
• 复杂参数传递失败：必须用States.JsonToString()将JSON对象转为字符串，否则EMR Serverless会将其解析为多个零散参数；
• 权限问题：确认Step Functions执行角色有调用emr-serverless:StartJobRun的权限，EMR Serverless执行角色有访问代码存储桶和相关资源的权限。

内容的提问来源于stack exchange，提问作者george-ognyanov