Воздушный поток Emr Dag прошел успешно, но кластер не запущен

Я пытаюсь запустить кластер 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)

Я пытался найти, например, или хороший пример, но на сайте воздушного потока не так много, кроме определения функции.

для создания потока заданий этот журнал повторяется несколько раз

для «добавить шаг» у меня есть это в журнале