title: Airflow EMR execute step from Sensor
body: <p>I made the following DAG in airflow where I am executing a set of EMRSteps to run my pipeline.</p>
  
  <pre><code>default_args = {
      'owner': 'airflow',
      'depends_on_past': False,
      'start_date': datetime(2017, 07, 20, 10, 00),
      'email': ['airflow@airflow.com'],
      'email_on_failure': False,
      'email_on_retry': False,
      'retries': 5,
      'retry_delay': timedelta(minutes=2),
  }
  
  dag = DAG('dag_import_match_hourly',
        default_args=default_args,
        description='Fancy Description',
        schedule_interval=timedelta(hours=1),
        dagrun_timeout=timedelta(hours=2))
  
  try:
      merge_s3_match_step = EmrAddStepsOperator(
          task_id='merge_s3_match_step',
          job_flow_id=cluster_id,
          aws_conn_id='aws_default',
          steps=create_step('Merge S3 Match'),
          dag=dag
      )
  
      mapreduce_step = EmrAddStepsOperator(
          task_id='mapreduce_match_step',
          job_flow_id=cluster_id,
          aws_conn_id='aws_default',
          steps=create_step('MapReduce Match Hourly'),
          dag=dag
      )
  
      merge_hdfs_step = EmrAddStepsOperator(
          task_id='merge_hdfs_step',
          job_flow_id=cluster_id,
          aws_conn_id='aws_default',
          steps=create_step('Merge HDFS Match Hourly'),
          dag=dag
      )
  
      ## Sensors
      check_merge_s3 = EmrStepSensor(
          task_id='watch_merge_s3',
          job_flow_id=cluster_id,
          step_id="{{ task_instance.xcom_pull('merge_s3_match_step', key='return_value')[0] }}",
          aws_conn_id='aws_default',
          dag=dag
      )
  
      check_mapreduce = EmrStepSensor(
          task_id='watch_mapreduce',
          job_flow_id=cluster_id,
          step_id="{{ task_instance.xcom_pull('mapreduce_match_step', key='return_value')[0] }}",
          aws_conn_id='aws_default',
          dag=dag
      )
  
      check_merge_hdfs = EmrStepSensor(
          task_id='watch_merge_hdfs',
          job_flow_id=cluster_id,
          step_id="{{ task_instance.xcom_pull('merge_hdfs_step', key='return_value')[0] }}",
          aws_conn_id='aws_default',
          dag=dag
      )
  
      mapreduce_step.set_upstream(merge_s3_match_step)
      merge_s3_match_step.set_downstream(check_merge_s3)
  
      mapreduce_step.set_downstream(check_mapreduce)
  
      merge_hdfs_step.set_upstream(mapreduce_step)
      merge_hdfs_step.set_downstream(check_merge_hdfs)
  
  except AirflowException as ae:
      print ae.message
  </code></pre>
  
  <p>The DAG works fine but <strong>I would like to use the Sensors to make sure that I am going to execute the next step if and only if the EMR Job has been completed correctly</strong>. I tried few things but none of them are working. The code above doesn't do the job properly. Does someone know how to use the EMRSensorStep to achieve my goal ?</p>