Mailing-List: contact dev-help@airflow.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@airflow.incubator.apache.org
Content-Type: text/plain; format=flowed
From: ziqr@vo.yoo.ro
To: dev@airflow.incubator.apache.org
Subject: Help reusing filepath from previous task
Content-Transfer-Encoding: 7bit
Date: Fri, 08 Jul 2016 17:58:39 +0000
Message-Id: <1468000719847-f9ae8a50-980aa15e-98a6dd93@vo.yoo.ro>
MIME-Version: 1.0
archived-at: Fri, 08 Jul 2016 17:58:33 -0000

Hello.

I'd like to reuse some info given by a task to feed subsequent S3KeySensor 
checks, but after trying several ideas I'm not able to make this work.

Basically my current DAG is as follow :

copy_data = BashOperator(
     task_id='copy_data',
    # this script copies data to S3
     bash_command='do_stuff.sh',
     dag=dag)

s3_path = xxxx
for data_type in ("fileA", "fileB", "fileC"):
     S3Sensor_task = S3KeySensor(
         task_id='check_' + data_type,
         poke_interval=20,
         timeout=60,
         retry_delay=timedelta(seconds=30),
         bucket_key=s3_path +  data_type,
         bucket_name='xxxx',
         s3_conn_id='s3_conn_id',
         dag=dag)
     S3Sensor_task.set_upstream(copy_data)

This works great, but I'd rather avoid duplicating the S3 path in the DAG 
and in the do_stuff.sh script.
So I tought pushing S3 path from the do_stuff.sh to xcom variables, but 
then I need access to thoses items in my for loop.
I've tried SubDag but once again I did not manage to read xcom.

Could someone give me some hints ?

Regards