Create Unique file name and access that file in all airflow task

user_1881997 0
Viewed 2498

Can we create unique file name every time airflow dag run and access that file from all tasks? I tried creating global variable (output_filename) and appended timestamp to it. But when ever i access that file_name in tasks,every task generates different file name as it is calculating timestamp in every task. Below is sample code:

table_name = 'Test_ABC'
start_date = datetime.now()
cur_tmpstp = start_date.strftime('%Y_%m_%d')

output_filename = table_name + "_" + cur_tmpstp + ".csv"
S3_landing_path = "s3://abc/"

def clean_up():
    if os.path.exists(output_filename):
        os.remove(output_filename)


task_1 = BashOperator(
    task_id='task_1',
    bash_command="aws s3 cp %s %s/ " %(output_filename, S3_landing_path, ),
    dag=dag)

task_2_cleanup = PythonOperator(
    task_id='task_2_cleanup',
    python_callable=clean_up,
    dag=dag)

We have more tasks where we have to access output_filename. How can we access output_filename global variable in all tasks?

apache-airflow airflow
2 Answers

If you need the timestamp with time granularity, it is possible to use global variables and a task with python operator:

DAG_NAME = 'Some DAG name'

ts = Variable.get(f"{DAG_NAME}_ts", default_var=None)

def generate_ts(*args, **kwargs):
    ts = datetime.now().isoformat()
    Variable.set(f"{DAG_NAME}_ts", ts)

generate_ts_task = PythonOperator(
    task_id='generate_ts',
    python_callable=generate_ts,
    dag=dag,
)
user_117436200
Related