在 AWS Step Functions 中传递变量

Question

我对 AWS Step Functions 和 AWS Lambda Functions 非常陌生，真的需要一些帮助。我有一个状态机，我试图检查某个文件是否存在于我的 S3 存储桶中，然后如果文件存在则让状态机遵循一个路径，如果不存在则让状态机遵循不同的路径。

下面显示了我的状态机代码的开头，涵盖了这个问题

{
  "Comment": "This is a test for running the structure of the CustomCreate job.",
  "StartAt": "PreStep",
  "States": {
    "PreStep": {
      "Comment": "Check that all the necessary files exist before running the job.",
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:XXXXXXXXXX:function:CustomCreate-PreStep-Function",
      "Next": "Run Job Choice"
    },
    "Run Job Choice": {
      "Comment": "This step chooses whether or not to go forward with running the main job.",
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.FoundNecessaryFiles",
          "BooleanEquals": true,
          "Next": "Spin Up Cluster"
        },
        {
          "Variable": "$.FoundNecessaryFiles",
          "BooleanEquals": false,
          "Next": "Do Not Run Job"
        }
      ]
    },
    "Do Not Run Job": {
      "Comment": "This step triggers if the PreStep fails and the job should not run.",
      "Type": "Fail",
      "Cause": "PreStep unsuccessful"
    },
    "Spin Up Cluster": {
      "Comment": "Spins up the EMR Cluster.",
      "Type": "Pass",
      "Next": "Update Env"
    },
    "Update Env": {
      "Comment": "Update the environment variables in the EMR Cluster.",
      "Type": "Pass",
      "Next": "Run Job"
    },
    "Run Job": {
      "Comment": "Add steps to the EMR Cluster.",
      "Type": "Pass",
      "End": true
    }
  }
}

以下代码描述了我的 CustomCreate-PreStep-Function Lambda 函数

exports.handler = async function(event, context, callback) {
     var AWS = require('aws-sdk');
     var s3 = new AWS.S3();
     var params = {Bucket: 'BUCKET_NAME', Key: 'FILE_NAME'};
     s3.getObject(params, function(err, data) {

        if (err) {
            console.log(err, err.stack);
            // file does not exist
            console.log("failed");
            callback(null,false);
        }
        else {
            console.log(data);
            //file exist
            console.log("succeeded");
            var FoundNecessaryFiles = true;
            // return FoundNecessaryFiles;
            callback(null,event.FoundNecessaryFiles=true);
        }    
    });
};

我尝试了很多方法，但都无法正常工作。如您所见，我正在尝试使用 Lambda 函数将状态为真/假的变量 FoundNecessaryFiles 传回，具体取决于是否找到了文件，并使用该值来指导在下一步。最好通过修复变量回传来解决这个问题，因为我稍后可能需要在状态机中再次使用该方法，但我也愿意接受另一种解决方案，无论是组合步骤还是其他任何可能的方法.

另外，如果存在正确的文件，我在此过程中的下一步将是启动一个 AWS EMR 集群，我也不清楚如何进行完成。如果有人能够在使用 Step Functions 运行 AWS EMR 集群方面提供任何帮助，我将不胜感激。

Answer 1

我解决了传递变量的最初问题，但是，我仍然可以通过 Step Functions 获得一些帮助，让 EMR 集群运行。

对于那些可能遇到类似问题的人，我通过将我的 Lambda 函数更改为以下解决了我的变量传递问题

exports.handler = function(event, context, callback) {
     var AWS = require('aws-sdk');
     var s3 = new AWS.S3();
     var params = {Bucket: 'BUCKET_NAME', Key: 'FILE_NAME'};
     s3.getObject(params, function(err, data) {

        if (err) {
            console.log(err, err.stack);
            // file does not exist
            console.log("failed");
            event.FoundNecessaryFiles = false;
            callback(null,event);
        }
        else {
            console.log(data);
            //file exist
            console.log("succeeded");
            event.FoundNecessaryFiles = true;
            callback(null,event);
        }    
    });
};

我的下一个问题是设置 AWS EMR 集群。我目前的第一个任务是启动 EMR 集群，这可以通过直接使用 Step Function JSON 来完成，或者最好使用我位于 S3 存储桶上的 JSON 集群配置文件来完成。我的下一个任务是更新 EMR 集群环境变量，我的 S3 存储桶上有一个 .sh 脚本可以执行此操作，我只是不知道如何使用 Step Functions 将其应用于 EMR 集群。我的第三个任务是向 EMR 集群添加一个包含 spark-submit 命令的步骤，该命令在我的 S3 存储桶上的 JSON 配置文件中进行了描述，可以以与上传环境配置类似的方式上传到 EMR 集群上一步中的文件。最后，我想要一个确保 EMR 集群在完成运行后终止的任务。

非常感谢您对此提供的任何帮助，无论您遵循我上面概述的结构还是知道改变结构的解决方案，我都很乐意提供任何帮助。

Answer 2

这可能对某些人有所帮助，因为 AWS 步骤函数中引入了新功能。

1. 为了在 EMR 集群上设置变量或部署软件，应该使用 BootStrap 操作。您可以提供引导脚本的 S3 路径。引导脚本作为集群创建过程的一部分运行。

2. 可以使用 Step Functions 添加 EMR 步骤（同步或异步）。可以将 spark-submit 构建为 Args 数组。这是一个例子：

ARGS_ARRAY = [ "spark-submit",
              "--master=yarn",
              "--deploy-mode=cluster",
              "--name=Generate Bulk File",
              "--driver-cores=1",
              "--driver-memory=2g",
              "--executor-cores=5",
              "--executor-memory=9g",
              "/home/hadoop/charu/mySparkJob.py", 
              input_file_name,
              output_file_name] 

1. 可以将工作流中的最后一步添加为“终止 EMR 集群”，具体取决于它之前的 EMR 步骤，即如果前一个 EMR 步骤是同步的并且成功完成，那么只有集群将被终止。 （假设这就是一个人想要的）