How can I pass SQL as a file w/ Airflow's Postgres Operator?

Airflow
1

To pass SQL as a file when leveraging the Postgres Operator you just have to provide a file name with .sql at the end and make sure it’s in your image.

You could consider:

  • Creating a sub folder in the dags folder named sql and putting all of your .sql files in there
  • Using jinja templates and macros in your .sql files too (if you wanted to pass in execution_date to filter your sql)

Example

DAG Code

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.postgres_operator import PostgresOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2019, 8, 25),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG('example_dag',
            max_active_runs=3,
            schedule_interval='@daily',
            default_args=default_args) as dag:

    t1 = PostgresOperator(
        task_id='my_task',
        sql='sql/my_query.sql'
    )

Astro-Airflow Directory


├── dags
│   ├── example-dag.py
│   └── sql
│       └── my_query.sql
├── include
├── packages.txt
├── plugins
│   └── example-plugin.py
└── requirements.txt
3 Likes
2

Another option if you want your sql folder to exist outside of the dag folder is to put the path into the arguments for your dag:

with DAG(‘example_dag’,
max_active_runs=3,
schedule_interval=‘@daily’,
template_searchpath=‘/usr/local/airflow/sql’,
default_args=default_args)

Then you can just reference the file name:

t1 = PostgresOperator(
task_id=‘my_task’,
sql=‘my_query.sql’
)

Directory Tree:

├── dags
│ ├── example-dag.py
├── sql
│ └── my_query.sql
├── include
├── packages.txt
├── plugins
│ └── example-plugin.py
└── requirements.txt

Also might be a good idea to use an airflow variable to store the path and just do:

template_searchpath=Variable.get(‘sql_path’)

2 Likes