Data Engineering Path · Airflow
Step Functions Operators & Hooks
A Second Orchestrator, on Purpose
It looks strange at first — Airflow orchestrating another orchestrator — but it's a genuinely common, deliberate pattern: Step Functions coordinating a tight sequence of native AWS service calls (fast, cheap, no Airflow worker sitting idle waiting), with Airflow as the higher-level scheduler that kicks it off, waits for it, and slots it into the rest of the pipeline.
Starting an Execution
from airflow.providers.amazon.aws.operators.step_function import StepFunctionStartExecutionOperator
from airflow.providers.amazon.aws.sensors.step_function import StepFunctionExecutionSensor
start_execution = StepFunctionStartExecutionOperator(
task_id="start_processing_workflow",
state_machine_arn="arn:aws:states:us-east-1:123456789012:stateMachine:order-processing",
state_machine_input={"date": "{{ ds }}", "source": "airflow"},
aws_conn_id="aws_default",
)
wait_for_completion = StepFunctionExecutionSensor(
task_id="wait_for_workflow_completion",
execution_arn=start_execution.output,
aws_conn_id="aws_default",
poke_interval=15,
)
start_execution >> wait_for_completion
Run for real, against a genuine (if intentionally minimal) Step Functions state machine:
Figure — start_execution.output (the execution ARN) flows directly into the sensor's execution_arn parameter via XCom, exactly as shown in the code above.
When This Pattern Beats Building the Same Logic as Airflow Tasks Directly
If a sub-workflow is a tight sequence of AWS API calls with conditional branching (e.g., "call this Lambda, if it returns X call this other Lambda, else call this one") that all needs to happen with sub-second handoffs, Step Functions runs that natively and cheaply. Modeling the same thing as individual Airflow tasks works too, but adds Airflow scheduling latency (seconds, not milliseconds) between every step - fine for most data pipelines, sometimes not fine for tightly-coupled event processing.
If a sub-workflow is a tight sequence of AWS API calls with conditional branching (e.g., "call this Lambda, if it returns X call this other Lambda, else call this one") that all needs to happen with sub-second handoffs, Step Functions runs that natively and cheaply. Modeling the same thing as individual Airflow tasks works too, but adds Airflow scheduling latency (seconds, not milliseconds) between every step - fine for most data pipelines, sometimes not fine for tightly-coupled event processing.
StepFunctionsHook Directly
from airflow.providers.amazon.aws.hooks.step_function import StepFunctionHook
def start_and_get_output(**context):
hook = StepFunctionHook(aws_conn_id="aws_default")
execution_arn = hook.start_execution(
state_machine_arn="arn:aws:states:us-east-1:123456789012:stateMachine:order-processing",
state_machine_input={"date": context["ds"]},
)
return execution_arn
Don't Duplicate Retry Logic
Step Functions has its own robust retry/catch semantics per state. Avoid wrapping
Step Functions has its own robust retry/catch semantics per state. Avoid wrapping
StepFunctionStartExecutionOperator in aggressive Airflow-level retries on top of that - if the state machine itself already retries internally, an Airflow-level retry on top just re-triggers the whole workflow from scratch instead of resuming where it left off.
arrow_back Previous Topic
ECS and Batch Operators and Hooks
Next Topic arrow_forward
KubernetesPodOperator