amazon-web-services - 在 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 函数传回状态为 true/false 的变量 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 集群在完成运行后终止。

如果您对此提供任何帮助，我们将不胜感激，无论您是遵循我上面概述的结构，还是知道改变结构的解决方案，我都很乐意提供任何帮助。