python - Airflow/Amazon EMR : The VPC/subnet configuration was invalid: Subnet is required : The specified instance type m5. xlarge 只能在VPC中使用

Question

我想创建一个通过 Amazon EMR 上的 Airflow 触发的 emr 集群。 emr 集群显示在 Amazon EMR 的 UI 中，但有一条错误消息:“VPC/子网配置无效:需要子网:指定的实例类型 m5.xlarge 只能在 VPC 中使用”

下面是 Airflow 脚本中使用的此任务的代码片段和 json 格式的配置详细信息。

我的问题是如何将有关 VPC 和子网的信息(id 代码)合并到 json 中(如果可能的话)？那里没有明确的例子。

提示:网络和 EC2 子网已经创建

JOB_FLOW_OVERRIDES = {
    "Name": "sentiment_analysis",
    "ReleaseLabel": "emr-5.33.0",
    "Applications": [{"Name": "Hadoop"}, {"Name": "Spark"}], # We want our EMR cluster to have HDFS and Spark
    "Configurations": [
        {
            "Classification": "spark-env",
            "Configurations": [
                {
                    "Classification": "export",
                    "Properties": {"PYSPARK_PYTHON": "/usr/bin/python3"}, # by default EMR uses py2, change it to py3
                }
            ],
        }
    ],
    "Instances": {
        "InstanceGroups": [
            {
                "Name": "Master node",
                "Market": "SPOT",
                "InstanceRole": "MASTER",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 1,
            },
            {
                "Name": "Core - 2",
                "Market": "SPOT", # Spot instances are a "use as available" instances
                "InstanceRole": "CORE",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 2,
            },
        ],
        "KeepJobFlowAliveWhenNoSteps": True,
        "TerminationProtected": False, # this lets us programmatically terminate the cluster
    },
    "JobFlowRole": "EMR_EC2_DefaultRole",
    "ServiceRole": "EMR_DefaultRole",
}

create_emr_cluster = EmrCreateJobFlowOperator(
    task_id="create_emr_cluster",
    job_flow_overrides=JOB_FLOW_OVERRIDES,
    aws_conn_id="aws_default",
    emr_conn_id="emr_default",
    dag=dag,
)

Answer 1

EmrCreateJobFlowOperator 从 emr.py 调用 create_job_flow与 boto3 emr client 中的相同 api 匹配.

因此，您可以将带有子网 ID 的项目 "Ec2SubnetId" 作为 "Instances" 字典中的值。

这对我适用于 Apache Airflow 2.0.2