Airflow - AWS Data Engineering

Airflow — Orchestration

Open-source workflow management for complex data pipelines — scheduling, monitoring, managing. From file: handles 10 CSV via SFTP + 2 REST APIs + 15 JDBC tables.

100GB daily example from file: MySQL + API → DMS incremental CDC → S3 → Lambda → Crawler (EventBridge) → Glue → S3 → Airflow (S3KeySensor → S3ToRedshiftOperator) → Redshift. Orchestrated via Airflow.

2GB MySQL example: DMS → S3 → Lambda (pandas) → S3 → Airflow/Step Functions.

Operators

Operator	Use
PythonOperator	Run Python functions, trigger processing
BashOperator	Run bash commands (ls, cd, mkdir, cp, mv, rm)
S3ToRedshiftOperator	Move S3 files to Redshift (COPY)
SqlOperator	Interact with SQL objects
EmailOperator	Send success/failure notifications
DummyOperator	Placeholder for start/end

Sensors

Sensor	Waits For
FileSensor	File to appear
HttpSensor	HTTP endpoint available
SqlSensor	SQL query to return desired result
S3KeySensor	Key to exist in S3 bucket

Example flow: dummy >> FileSensor(True) >> EmailOperator >> PythonOperator(df.write) >> S3KeySensor(True) >> S3ToRedshiftOperator >> dummy

DAG Example (from file)

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from datetime import datetime
import pandas as pd, pysftp, requests, json

default_args = {'owner':'airflow','depends_on_past':True,'start_date':datetime(2026,3,25),'retries':3}
dag = DAG(dag_id='my_dag', default_args=default_args, schedule_interval='@daily', catchup=False)

def extract_from_sftp(**kwargs):
    with pysftp.Connection(sftp_host, port=22, username='user', password='pass') as sftp:
        sftp.get('/remote/file.csv', 'tmp/data.csv')

def extract_from_jdbc(**kwargs):
    from airflow.providers.postgres.hooks.postgres import PostgresHook
    hook = PostgresHook(postgres_conn_id='jdbc_conn')
    df = hook.get_pandas_df('select * from mytable')
    df.to_csv('tmp/data.csv', index=False)

extract_sftp = PythonOperator(task_id='extract_from_sftp', python_callable=extract_from_sftp, dag=dag)
extract_jdbc = PythonOperator(task_id='extract_from_jdbc', python_callable=extract_from_jdbc, dag=dag)
sensor = S3KeySensor(task_id='s3_key_sensor', bucket_name='mybucket', bucket_key='finaldata/data.csv', poke_interval=60, timeout=3600, dag=dag)
load = S3ToRedshiftOperator(task_id='load_to_redshift', schema='public', table='mytable', s3_bucket='mybucket', s3_key='finaldata/data.csv', redshift_conn_id='redshift', dag=dag)

extract_sftp >> sensor
extract_jdbc >> sensor
sensor >> load

Hooks: PostgresHook, MySqlHook, OracleHook, JdbcHook, HttpHook, S3Hook. XCom: xcom_push/xcom_pull to exchange data between tasks. DAG = collection of tasks with >> dependencies.

Variables/Connections stored in Airflow Admin → Variables/Connections (e.g., s3_key, SFTP/JDBC/AWS).

Catchup & DependsOnPast

Combo	Behavior	Use When
catchup=True, depends=True	Creates many past runs, each waits for previous — sensitive, one failure blocks all	Strict order, monitoring retries
catchup=False, depends=True	No backfill, today waits for yesterday	Daily incremental, CDC — most common in production
catchup=True, depends=False	Backfills, no waiting	Full refresh, historical rebuild
catchup=False, depends=False	Only latest, no dependency	Daily dashboards, API daily calls

From file: DAG daily, start_date 25032025, catchup True creates runs for past dates, False only for current/future. DependsOnPast = task waits for previous run's same task success.

Airflow vs Step Functions

Step Functions	Airflow
AWS managed, serverless, event-driven, short orchestration, pay per transaction, built-in retries, max 1 year (15 min Lambda)	Open source, self-managed (or MWAA), perfect for ETL/complex dependencies, free (pay for server), configurable retries, no limit, parallel via XCom
Easy setup, stateless, no parallel XCom	Complex setup, parallel executions, XCom push/pull