如何按需创建 EMR 集群并执行 aws emr 命令？答案

【问题标题】：How to create EMR cluster on demand and execute aws emr command?如何按需创建 EMR 集群并执行 aws emr 命令？
【发布时间】：2018-07-13 06:18:48
【问题描述】：

我想按需执行 Spark 作业。因此，只有当我收到触发事件时，我才想使用随此触发事件到达的输入来执行 Spark 作业。由于触发事件不频繁，我不想使用 Spark Streaming。我的目标是在 AWS EMR 集群中部署该工具。我希望能够按需（通过触发器）创建 EMR 集群，在那里执行 Spark 作业并关闭集群。有没有很好的例子说明如何从 Scala 处理这些操作？

【问题讨论】：

这取决于，什么样的事件会触发？你的问题太模糊了，但总的来说，不仅仅是一种语言，要对事件执行操作，你需要某种消息队列。例如，您可以混合使用 AWS CloudWatch（用于事件）和 AWS Data Pipeline（用于 Spark 作业）
通过 Amazon 控制台手动设置 EMR。测量集群上线需要多长时间。然后从 AWS CLI 执行相同的操作。您会发现 EMR 设置需要一段时间（超过几分钟）。还要注意启动集群的最低账单费率。与 Lambda、EC2 等相比，EMR 不是按需快速启动和停止类型的服务。不要使用 CloudWatch 启动 EMR。您最终可能会在云中使用大量昂贵的集群。
@RobertoCongiu：触发器（事件）将来自 RabbitMQ。我收到它们并在 Scala 中处理它们。但我的问题是如何从 Scala 运行 launch aws emr 命令以创建集群并运行 Spark 作业？
@JohnHanley：最接近我需要的实现是使用AmazonElasticMapReduceClient:docs.aws.amazon.com/emr/latest/ReleaseGuide/…

标签： amazon-web-services apache-spark emr amazon-emr

【解决方案1】：

AWS Data Pipeline 似乎是您定义的问题的正确解决方案。 AWS Data Pipeline 允许您在 AWS 基础设施中连接多种服务，例如存储和处理。
您可以使用 AWS 数据管道中的 EMRActivity 创建 EMR 作业。管道将在满足先决条件或按预定时间间隔触发。
它将使用您指定的规范和您定义的 Spark 步骤设置 EMR 集群
作业完成后可以自动终止集群。

这个question SO 将帮助您入门。

您还可以在使用 Choose a Template 选项创建管道时启动 AWS 数据管道 using this definition。对于此选项，您可以使用上面共享的模板。

【讨论】：

【解决方案2】：

只要允许启动 EMR 集群，Lambda 函数可能是一个很好的解决方案。 Lambda 函数可以按需启动，也可以使用多个不同的触发器来调用。

这可能是开始设置 Lambda 函数的好模板：

import sys
import time

import boto3

def lambda_handler(event, context):
    conn = boto3.client("emr")
    # chooses the first cluster which is Running or Waiting
    # possibly can also choose by name or already have the cluster id
    clusters = conn.list_clusters()
    # choose the correct cluster
    clusters = [c["Id"] for c in clusters["Clusters"] 
                if c["Status"]["State"] in ["RUNNING", "WAITING"]]
    if not clusters:
        sys.stderr.write("No valid clusters\n")
        sys.stderr.exit()
    # take the first relevant cluster
    cluster_id = clusters[0]
    # code location on your emr master node
    CODE_DIR = "/home/hadoop/code/"

    # spark configuration example
    step_args = ["/usr/bin/spark-submit", "--spark-conf", "your-configuration",
                 CODE_DIR + "your_file.py", '--your-parameters', 'parameters']

    step = {"Name": "what_you_do-" + time.strftime("%Y%m%d-%H:%M"),
            'ActionOnFailure': 'CONTINUE',
            'HadoopJarStep': {
                'Jar': 's3n://elasticmapreduce/libs/script-runner/script-runner.jar',
                'Args': step_args
            }
        }
    action = conn.add_job_flow_steps(JobFlowId=cluster_id, Steps=[step])
    return "Added step: %s"%(action)

【讨论】：

【解决方案3】：

与其通过 api 旋转 EMR 集群，不如拥有 EMR 在成云模板中配置并开始成云模板 (CFT)。通过 CFT，您可以处理 IAM 角色，创建 S3 水桶等
通过 lambda 旋转 EMR 集群；如果有任何火花作业正在进行或等待，则让另一个 lambda 监视 EMR。如果在 空闲时间（比如 15 分钟） 内没有，则在 CFT 上发出拆解，进而终止 EMR

【讨论】：