amazon-emr - 有没有一种方法可以将步函数输入值输入到EMR步骤Args中

Question

我们正在使用AWS EMR集群运行批处理 Spark 作业。这些作业会定期运行，我们希望通过AWS Step Functions进行整理。

从2019年11月开始，Steps本地支持EMR。将步骤添加到集群时，我们可以使用以下配置:

"Some Step": {
    "Type": "Task",
    "Resource": "arn:aws:states:::elasticmapreduce:addStep.sync",
    "Parameters": {
        "ClusterId.$": "$.cluster.ClusterId",
        "Step": {
            "Name": "FirstStep",
            "ActionOnFailure": "CONTINUE",
            "HadoopJarStep": {
                "Jar": "command-runner.jar",
                "Args": [
                    "spark-submit",
                    "--class",
                    "com.some.package.Class",
                    "JarUri",
                    "--startDate",
                    "$.time",
                    "--daysToLookBack",
                    "$.daysToLookBack"
                ]
             }
         }
     },
     "Retry" : [
         {
             "ErrorEquals": [ "States.ALL" ],
             "IntervalSeconds": 1,
             "MaxAttempts": 1,
             "BackoffRate": 2.0
         }
     ],
     "ResultPath": "$.firstStep",
     "End": true
}

在HadoopJarStep的Args列表中，我们想动态设置参数。例如如果状态机执行的输入是:

{
    "time": "2020-01-08",
    "daysToLookBack": 2
}

配置中以“$”开头的字符串。在执行状态机时应相应地对其进行替换，并且EMR集群上的步骤应运行 command-runner.jar spark-submit --class com.some.package.Class JarUri --startDate 2020-01-08 --daysToLookBack 2。但是相反，它运行 command-runner.jar spark-submit --class com.some.package.Class JarUri --startDate $.time --daysToLookBack $.daysToLookBack。

有谁知道有没有办法做到这一点？

Answer 1

参数允许您定义键值对，因此“Args”键的值是一个数组，因此您将无法动态引用该数组中的特定元素，而需要引用整个数组。例如“Args。$”:“$。Input.ArgsArray”。

因此，对于您的用例，实现此目的的最佳方法是在调用此状态之前添加预处理状态。在预处理状态下，您可以调用Lambda函数并通过代码设置输入/输出的格式，也可以使用Pass State重新格式化数据，然后在任务中使用Stated Parameters来简单地为数组添加动态值。您可以使用JSONPath来获取您在预处理器中定义的数组。这是一个例子:

{
"Comment": "A Hello World example of the Amazon States Language using Pass states",
"StartAt": "HardCodedInputs",
"States": {
    "HardCodedInputs": {
        "Type": "Pass",
        "Parameters": {
            "cluster": {
                "ClusterId": "ValueForClusterIdVariable"
            },
            "time": "ValueForTimeVariable",
            "daysToLookBack": "ValueFordaysToLookBackVariable"
        },
        "Next": "Pre-Process"
    },
    "Pre-Process": {
        "Type": "Pass",
        "Parameters": {
            "FormattedInputsForEmr": {
                "ClusterId.$": "$.cluster.ClusterId",
                "Args": [
                    {
                        "Arg1": "spark-submit"
                    },
                    {
                        "Arg2": "--class"
                    },
                    {
                        "Arg3": "com.some.package.Class"
                    },
                    {
                        "Arg4": "JarUri"
                    },
                    {
                        "Arg5": "--startDate"
                    },
                    {
                        "Arg6.$": "$.time"
                    },
                    {
                        "Arg7": "--daysToLookBack"
                    },
                    {
                        "Arg8.$": "$.daysToLookBack"
                    }
                ]
            }
        },
        "Next": "Some Step"
    },
    "Some Step": {
        "Type": "Pass",
        "Parameters": {
            "ClusterId.$": "$.FormattedInputsForEmr.ClusterId",
            "Step": {
                "Name": "FirstStep",
                "ActionOnFailure": "CONTINUE",
                "HadoopJarStep": {
                    "Jar": "command-runner.jar",
                    "Args.$": "$.FormattedInputsForEmr.Args[*][*]"
                }
            }
        },
        "End": true
    }
  }
}