amazon-web-services - Airflow Emr Dag 成功但集群未启动

Question

我正在尝试启动一个 AWS emr 集群并使用 EmrCreateJobFlowOperator 和 EmrAddStepsOperator 提交一个步骤，我的两个步骤都成功了但是集群永远不会启动，即使没有步骤

这两个步骤都变为成功状态

这是我的代码

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'retry_delay': timedelta(minutes=2),
    'start_date': datetime(2019, 1, 1),
    'end_date': datetime(2019, 2, 1),

    'depends_on_past': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)    
}


step_args = ["spark-submit", '../test.py']

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
            }
        }]



JOB_FLOW_OVERRIDES = {
'Instances': {
    'InstanceGroups': [
        {

            'InstanceRole': 'MASTER',
            'InstanceType': 'm4.large',
            'InstanceCount': 1
        },
        {

            'InstanceRole': 'CORE',
            'InstanceType': 'm4.large',
            'InstanceCount': 2,

        }
    ]},
    'Name':'airflow-monthly_agg_custom',

            'BootstrapActions':[{
            'Name': 'Install',
            'ScriptBootstrapAction': {
                'Path': 's3://dep-buck/bootstrap.sh'
            }
        }],
    'Configurations': [
  {
     "Classification": "spark-env",
     "Configurations": [
       {
         "Classification": "export",
         "Properties": {
            "PYSPARK_PYTHON": "/usr/bin/python3"
          }
       }
    ]
  }
]}

dag = DAG('emr_job_flow_automatic_steps_7',
          default_args=default_args,
          schedule_interval="@daily",
          max_active_runs=1,

          #    schedule_interval='*/1 * * * *',

          catchup=True,

          #         dagrun_timeout=timedelta(seconds=10)
          )

cluster_creator = EmrCreateJobFlowOperator(
    task_id='create_job_flow2',
    job_flow_overrides=JOB_FLOW_OVERRIDES,
    aws_conn_id='aws_default',
    emr_conn_id='emr_default',
    dag=dag
)

step_adder = EmrAddStepsOperator(
    task_id='add_steps',
    job_flow_id="{{ task_instance.xcom_pull('create_job_flow2', key='return_value') }}",
    aws_conn_id='aws_default',
    steps=step,
    dag=dag
)

cluster_creator.set_downstream(step_adder)

我曾尝试搜索示例或好的文档，但除了 Airflow 网站上的功能定义外，没有太多内容

为了创建作业流程，我将此日志重复了几次

对于“添加步骤”，我在日志中有这个

Answer 1

问题主要是关于用户和区域的可见性，它在默认区域启动集群，所以我不得不更改下面的属性

Airflow UI > admin > connection > aws_default > extra

{"region_name": "the region i was watching the ec2 console"}

Airflow UI > admin > connection > emr_default > extra

"VisibleToAllUsers": true,