How to get the uniform current date time throughout each loop in Airflow?

user_13516482 0
Viewed 1023

Below is a format of the DAG used, I am using for loop over input files in the DAG and rename the output files using the

CurrentDateTime = datetime.datetime.today().strftime("%Y%m%d%H%M%S")

#if currentDateTIme = 20210406010203

outputfile = ''
outputfile1 = ''
outputfile2 = ''
inputfiles = ['input1', 'input2']

API_CALL_TASK1 = {
source : inputfile1
filename : outputfile1 #20210406010513
}

API_CALL_TASK2 = {
source : inputfile2
filename : outputfile2 
}

for file in inputfiles:
    if file == 'input1'
        outputfile1 = f'inputFileName_{CurrentDateTime}' #20210406010303
        outputfile = f'inputFileName_{CurrentDateTime}' #20210406010303
    if file == 'input2'
        outputfile2 = f'inputFileName_{CurrentDateTime}' 
        outputfile = f'inputFileName_{CurrentDateTime}'        
    MOVE_OUTPUT_TO_BUCKET_TASK = (
                                 filename = f{outputfile} #20210406010423
                                 )

MOVE_OUTPUT_TO_BUCKET_TASK >> API_CALL_TASK1 >> API_CALL_TASK1

Here in the tasks - API_CALL_TASK1, API_CALL_TASK2 and to the MOVE_OUTPUT_TO_BUCKET_TASK the datetime in the filename is different as there is difference in time when each task is triggered.

How to get the uniform datetime for each file?

I want to pass the same filename from the loop to the MOVE_OUTPUT_TO_BUCKET_TASK and API_CALL_TASK1 or API_CALL_TASK2

google-cloud-composer airflow google-cloud-storage datetime python
1 Answers

As I understand you want to pass the same datetime in the format YYYYMMDD for both outputs in the loop.

Airflow provides us a set of Default Variables, which can be used across all templates. So you can use it to pass a constant value throughout your template. In your case, I see that you want the execution date to be the suffix of your output. Thus, you should use ds_nodash, as per documentation, it retrieves the execution date as YYYYMMDD.

In case you are using a Python operator with **kwargs, you can access it as kwargs['ds_nodash'].

user_125713870
Related