title: Airflow pass parameters to dependent task
body: <p>What is the way to pass parameter into dependent tasks in Airflow? I have a lot of bashes files, and i'm trying to migrate this approach to airflow, but i don't know how to pass some properties between tasks.</p>
  <p>This is a real example:</p>
  <pre><code>#sqoop bash template
  sqoop_template = &quot;&quot;&quot;
          sqoop job --exec {{params.job}} -- --target-dir {{params.dir}} --outdir /src/
      &quot;&quot;&quot;
  
  s3_template = &quot;&quot;&quot;
          s3-dist-cp --src= {{params.dir}} --dest={{params.s3}}
      &quot;&quot;&quot;
  
  
  
  #Task of extraction in EMR
  t1 = BashOperator(
          task_id='extract_account', 
          bash_command=sqoop_template, 
          params={'job': 'job', 'dir': 'hdfs:///account/' + time.now().strftime(&quot;%Y-%m-%d-%H-%M-%S&quot;)},
          dag=dag)
  #Task to upload in s3 backup.
  t2 = BashOperator(
          task_id='s3_upload',
          bash_command=s3_template,
          params={}, #here i need the dir name created in t1
          depends_on_past=True
      )
  
  t2.set_upstream(t1)
  </code></pre>
  <p>In t2 i need to access the dir name created in t1.</p>
  <h1>Solution</h1>
  <pre><code>#Execute a valid job sqoop
  def sqoop_import(table_name, job_name):
      s3, hdfs = dirpath(table_name)
      sqoop_job = job_default_config(job_name, hdfs)
      #call(sqoop_job)
      return {'hdfs_dir': hdfs, 's3_dir': s3}
  
  def s3_upload(**context):
      hdfs = context['task_instance'].xcom_pull(task_ids='sqoop_import')['hdfs_dir']
      s3 = context['task_instance'].xcom_pull(task_ids='sqoop_import')['s3_dir']
      s3_cpdist_job = [&quot;s3-dist-cp&quot;, &quot;--src=%s&quot; % (hdfs), &quot;--dest=%s&quot; % (s3)]
      #call(s3_cpdist_job)
      return {'s3_dir': s3} #context['task_instance'].xcom_pull(task_ids='sqoop_import')
  
  def sns_notify(**context):
      s3 = context['task_instance'].xcom_pull(task_ids='distcp_s3')['s3_dir']
      client = boto3.client('sns')
      arn = 'arn:aws:sns:us-east-1:744617668409:pipeline-notification-stg'
      response = client.publish(TargetArn=arn, Message=s3)
      return response
  
      
  </code></pre>
  <p>That's not is the definitive solution, so improvements are welcome. Thanks.</p>