airflow-tutorial-gist-06.py

# A minimalist idempotent aggregation query for clickstream data
aggregation_query_template = '''
BEGIN;

DELETE FROM clickstream_aggregated
WHERE click_date = '{{ execution_date }}'::DATE;

INSERT INTO clickstream_aggregated
SELECT click_timestamp::DATE AS click_date,
    SUM(CASE WHEN is_ad_display_event THEN 1 ELSE 0 END) AS nb_ad_display_events,
    SUM(CASE WHEN is_ad_search_event THEN 1 ELSE 0 END) AS nb_ad_search_events
FROM clickstream
WHERE click_timestamp::DATE = '{{ execution_date }}'::DATE
GROUP BY 1;

COMMIT;
'''

# A simple postgres SQL task execution
aggregation_task = PostgresOperator(
    task_id='aggregate_clickstream_data',
    postgres_conn_id=target_db,
    sql=aggregation_query_template,
    autocommit=False,
    dag=dag)

aggregation_task.set_upstream(sensor_task)