How do I repeat BigQueryOperator Dag and pass different dates to my sql file

Question

I have a query I want to run using the BigQueryOperator. Each day, it will run for the past 21 days. The sql file stays the same, but the date passed to the file changes. So for example today it will run for today's date, then repeat for yesterday's date, and then repeat for 2 days ago, all the way up to 21 days ago. So it will run on 7/14/2021, and so I need to pass this date to my sql file. Then It will run for 7/13/2021, and the date I need to pass to my sql file is 7/13/2021. How can I have this dag repeat for a date range, and dynamically pass this date to the sql file.

In the BigQueryOperator, variables are passed in the "user_defined_macros, section, so I don't know how to change the date I am passing. I thought about looping over an array of dates, but I don't know how to pass that date to the sql file linked in the BigQueryOperator.

My sql file is 300 lines long, so I included a simple example below, as people seem to ask for one.

DAG

with DAG(
    dag_id,
    schedule_interval='0 12 * * *',
    start_date=datetime(2021, 1, 1),
    template_searchpath='/opt/airflow/dags',
    catchup=False,
    user_defined_macros={"varsToPass":Var1
    }

) as dag:
    query_one = BigQueryOperator(
        task_id='query_one',
        sql='/sql/something.sql',
        use_legacy_sql=False,
        destination_dataset_table ='table',
        write_disposition = 'WRITE_TRUNCATE'
        
    )

sql file

SELECT * FROM table WHERE date = {{CHANGING_DATE}}

user_14624409 · Answer

Your code is confusing because you describe a repeated pattern of today,today-1 day, ..., today - 21 days however your code shows write_disposition = 'WRITE_TRUNCATE' which means that only the LAST query matters because each query erase the result of the previous one. Since no more information provided I assume you actually mean to run a single query between the today to today - 21 days. Also You didn't mention if the date that you are referring to is Airflow execution_date or today date.

If it's execution_date you don't need to pass any parameters. the SQL needs to be:

SELECT * FROM table WHERE date BETWEEN {{ execution_date }} AND
{{ execution_date - macros.timedelta(days=21) }}

If it's today then you need to pass parameter with params:

from datetime import datetime
query_one = BigQueryOperator(
    task_id='query_one',
    sql='/sql/something.sql',
    use_legacy_sql=False,
    destination_dataset_table ='table',
    write_disposition = 'WRITE_TRUNCATE',
    params={
            "end": datetime.utcnow().strftime('%Y-%m-%d'),
            "start": (datetime.now() - datetime.timedelta(days=21)).strftime('%Y-%m-%d')
    }
    
)

Then in the SQL you can use it as:

SELECT * FROM table WHERE date BETWEEN {{ params.start }} AND
{{ params.end }}

I'd like to point that if you are not using execution_date then I don't see the value of passing the date from Airflow. You can just do it directly with BigQuery by setting the query to:

SELECT *
FROM table
WHERE date BETWEEN DATE_SUB(current_date(), INTERVAL 21 DAY) AND current_date()

If my assumption was incorrect and you want to run 21 queries then you can do that with a loop as you described:

from datetime import datetime, timedelta
from airflow.contrib.operators.bigquery_operator import BigQueryOperator
a = []
for i in range(0, 21):
    a.append(
        BigQueryOperator(
            task_id=f'query_{i}',
            sql='/sql/something.sql',
            use_legacy_sql=False,
            destination_dataset_table='table',
            write_disposition='WRITE_TRUNCATE',  # This is probably wrong, I just copied it from your code.
            params={
                "date_value": (datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d')
            }
        )

    )
    if i not in [0]:
        a[i - 1] >> a[i]

Then in your /sql/something.sql the query should be:

SELECT * FROM table WHERE date = {{ params.date_value }}

As mentioned this will create a workflow :

Note also that BigQueryOperator is deprecated. You should use BigQueryExecuteQueryOperator which available in Google provider via

from airflow.providers.google.cloud.operators.bigquery import BigQueryExecuteQueryOperator

for more information about how to install Google provider please see the 2nd part of the following answer.

How do I repeat BigQueryOperator Dag and pass different dates to my sql file

1 Answers