如何在AWS Step Functions中动态设置EMR步骤的前一日日期参数

要实现执行Step Function时自动将date_path设为执行当日的前一日，可利用AWS Step Functions的内置日期处理函数完成，无需额外编写Lambda函数。具体修改方式如下：

核心修改逻辑 #

通过States.SubtractDays从执行起始时间（$$.Execution.StartTime）减去1天得到前一日日期，再用States.Format将日期格式化为year=YYYY/month=MM/day=DD/的字符串格式，替换原有的固定日期值。

修改后的完整状态机代码 #

{
  "Comment": "A description of my state machine",
  "StartAt": "EMR CreateCluster",
  "States": {
    "EMR CreateCluster": {
      "Type": "Task",
      "Resource": "arn:aws:states:::elasticmapreduce:createCluster.sync",
      "Parameters": {
        "Name": "IOretrieve",
        "ServiceRole": "EMR_DefaultRole",
        "JobFlowRole": "EMR_EC2_DefaultRole",
        "ReleaseLabel": "emr-6.8.0",
        "Applications": [
          {
            "Name": "Spark"
          }
        ],
        "LogUri": "s3://",
        "VisibleToAllUsers": true,
        "Instances": {
          "Ec2SubnetId": "subnet",
          "Ec2KeyName": "",
          "EmrManagedMasterSecurityGroup": "",
          "EmrManagedSlaveSecurityGroup": "",
          "KeepJobFlowAliveWhenNoSteps": true,
          "InstanceFleets": [
            {
              "InstanceFleetType": "MASTER",
              "Name": "Master",
              "TargetOnDemandCapacity": 1,
              "InstanceTypeConfigs": [
                {
                  "InstanceType": "m5.xlarge"
                }
              ]
            },
            {
              "InstanceFleetType": "CORE",
              "Name": "CORE",
              "TargetOnDemandCapacity": 5,
              "InstanceTypeConfigs": [
                {
                  "InstanceType": "r5.2xlarge"
                }
              ]
            }
          ]
        },
        "BootstrapActions": [
          {
            "Name": "Custom action",
            "ScriptBootstrapAction": {
              "Path": "s3://",
              "Args": []
            }
          }
        ],
        "Configurations": [
          {
            "Classification": "core-site",
            "Properties": {
              "fs.s3a.connection.maximum": "1000"
            }
          },
          {
            "Classification": "spark",
            "Properties": {
              "maximizeResourceAllocation": "true"
            }
          }
        ]
      },
      "ResultPath": "$.cluster",
      "Next": "Run first step"
    },
    "Run first step": {
      "Type": "Task",
      "Resource": "arn:aws:states:::elasticmapreduce:addStep.sync",
      "Parameters": {
        "ClusterId.$": "$.cluster.ClusterId",
        "Step": {
          "Name": "My first EMR step",
          "HadoopJarStep": {
            "Jar": "command-runner.jar",
            "Args": [
              "spark-submit",
              "--deploy-mode",
              "client",
              "s3://",
              "--local_run",
              "False",
              "--date_path",
              "States.Format('year={}/month={:02d}/day={:02d}/', States.SubtractDays($$.Execution.StartTime, 1).year, States.SubtractDays($$.Execution.StartTime, 1).month, States.SubtractDays($$.Execution.StartTime, 1).day)"
            ]
          }
        }
      },
      "ResultPath": "$.firstStep",
      "Next": "Run second step"
    },
    "Run second step": {
      "Type": "Task",
      "Resource": "arn:aws:states:::elasticmapreduce:addStep.sync",
      "Parameters": {
        "ClusterId.$": "$.cluster.ClusterId",
        "Step": {
          "Name": "My second EMR step",
          "HadoopJarStep": {
            "Jar": "command-runner.jar",
            "Args": [
              "spark-submit",
              "--deploy-mode",
              "client",
              "s3://",
              "--local_run",
              "False",
              "--date_path",
              "States.Format('year={}/month={:02d}/day={:02d}/', States.SubtractDays($$.Execution.StartTime, 1).year, States.SubtractDays($$.Execution.StartTime, 1).month, States.SubtractDays($$.Execution.StartTime, 1).day)"
            ]
          }
        }
      },
      "ResultPath": "$.secondStep",
      "Next": "EMR TerminateCluster"
    },
    "EMR TerminateCluster": {
      "Type": "Task",
      "Resource": "arn:aws:states:::elasticmapreduce:terminateCluster",
      "Parameters": {
        "ClusterId.$": "$.cluster.ClusterId"
      },
      "End": true
    }
  }
}

关键修改点说明 #

• States.SubtractDays($$.Execution.StartTime, 1)：获取Step Function执行起始时间的前一天日期对象
• States.Format：将日期对象的年、月、日格式化为指定字符串，其中{:02d}确保月份和日期始终为两位数字（如6月显示为06，5日显示为05）
• 替换了Run first step和Run second step中Args数组里的固定日期字符串，改为动态生成的日期路径

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