Implementing Airflow with real scripts

Question

I set up an Airflow server successfully. I want to run some test jobs but I am having trouble finding beginner guides which fit into what I am trying to do.

Current status:

• Python scripts to download files from SFTP (any file which does not exist on local machine) or create a file from a queryout
• Pandas scripts to read the data into memory, modify it in some way to prepare it for the database (look for new dimensions, remap, add calculations). Load data to appropriate table in database. Send email summaries (pandas to_html)

The logic I have for most of my scripts is based on if the file has not been processed, then process it. 'Processed' files are either organized by filename in a db table, or I move the file to a special processed folder.

The other logic I have is based on the date in the filename. I compare the dates of files which exist versus dates which should exist (a range of dates). If the file does not exist, then I create it (usually a BCP or PSQL query).

Do I just have Airflow run these .py files? Or should I alter my scripts to use some of the Airflow parameters/jinja templating?

I almost feel like I could use the BashOperator for almost everything. Would this work

dag_input = sys.argv[1]

def alter_table(query, engine=pg_engine):
    fake_conn = engine.raw_connection()
    fake_cur = fake_conn.cursor()
    fake_cur.execute(query)
    fake_conn.commit()
    fake_cur.close()

query_list = [
              f'SELECT * from table_1 where report_date = \'{dag_input}\'',
              f'SELECT * from table_2 where report_date = \'{dag_input}\'',
]

for value in query_list:
    alter_table(value)

Then the dag would be something like this, with an airflow parameter used for the sys.argv?

templated_command = """
        python download_raw.py "{{ ds }}"
"""

t3 = BashOperator(
    task_id='download_raw',
    bash_command=templated_command,
    dag=dag)

Answer 1

Since the code for this task is in python, I would use a PythonOperator.

Put a method in download_raw.py that takes **kwargs as parameters and you have access to everything in the context.

from download_raw import my_func

t3 = PythonOperator(
    task_id='download_raw',
    python_callable=my_func,
    dag=dag)


#inside download_raw.py

def my_func(**kwargs):
    context = kwargs
    ds = context['ds']
    ... (do your logic here)

I would do it like this or your bash command could get hideous when several pieces of the context.