AWS Step Functions for Data Engineers

Standard vs Express, state types, a full ETL state machine, .sync and callback integrations, data passing, Map and Distributed Map, error handling, triggering and monitoring.

1. What AWS Step Functions Is

AWS Step Functions orchestrates workflows as state machines defined in JSON (Amazon States Language, ASL). It calls AWS services directly (Glue, EMR, Athena, Lambda, ECS, Batch, SNS, DynamoDB and 200+ others), waits for them, retries failures, and shows each execution visually. It is serverless: no scheduler to run.

2. State Types

3. Example: A Daily ETL Pipeline

Validate the file with Lambda, run a Glue job and wait for it, check data quality, refresh Athena partitions, and notify. Every failure goes to one alert state.

{
  "Comment": "Daily transactions load",
  "StartAt": "ValidateFile",
  "States": {
    "ValidateFile": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": { "FunctionName": "file-validator", "Payload.$": "$" },
      "ResultSelector": { "rows.$": "$.Payload.rows", "valid.$": "$.Payload.valid" },
      "ResultPath": "$.validation",
      "Next": "IsValid",
      "Catch": [{ "ErrorEquals": ["States.ALL"], "ResultPath": "$.error", "Next": "NotifyFailure" }]
    },
    "IsValid": {
      "Type": "Choice",
      "Choices": [{ "Variable": "$.validation.valid", "BooleanEquals": true, "Next": "RunGlueJob" }],
      "Default": "NotifyFailure"
    },
    "RunGlueJob": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startJobRun.sync",
      "Parameters": {
        "JobName": "raw-to-silver-transactions",
        "Arguments": { "--run_date.$": "$.run_date" }
      },
      "ResultPath": "$.glue",
      "Retry": [{
        "ErrorEquals": ["Glue.ConcurrentRunsExceededException", "States.TaskFailed"],
        "IntervalSeconds": 60, "MaxAttempts": 2, "BackoffRate": 2.0
      }],
      "Catch": [{ "ErrorEquals": ["States.ALL"], "ResultPath": "$.error", "Next": "NotifyFailure" }],
      "Next": "AddPartition"
    },
    "AddPartition": {
      "Type": "Task",
      "Resource": "arn:aws:states:::athena:startQueryExecution.sync",
      "Parameters": {
        "QueryString.$": "States.Format('ALTER TABLE silver.transactions ADD IF NOT EXISTS PARTITION (dt = \\'{}\\')', $.run_date)",
        "WorkGroup": "data-eng"
      },
      "ResultPath": null,
      "Next": "NotifySuccess"
    },
    "NotifySuccess": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:ap-south-1:111122223333:pipeline-events",
        "Message.$": "States.Format('Loaded {} rows for {}', $.validation.rows, $.run_date)"
      },
      "End": true
    },
    "NotifyFailure": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:ap-south-1:111122223333:pipeline-alerts",
        "Message.$": "States.JsonToString($)"
      },
      "Next": "Failed"
    },
    "Failed": { "Type": "Fail", "Error": "PipelineFailed" }
  }
}

4. Service Integration Patterns

# a worker (or approval UI) completes a waiting callback task
import boto3, json
sfn = boto3.client("stepfunctions")
sfn.send_task_success(taskToken=token, output=json.dumps({"approved": True, "by": "risk-team"}))
# or: sfn.send_task_failure(taskToken=token, error="Rejected", cause="Totals do not reconcile")

5. Passing Data Between States

With the classic JSONPath query language, each state filters its input and output in a fixed order:

state input
  → InputPath        select part of the input           "$.order"
  → Parameters       build the API request              {"JobName": "x", "Arguments": {"--id.$": "$.id"}}
  → (task runs)
  → ResultSelector   keep only useful result fields     {"status.$": "$.JobRunState"}
  → ResultPath       where to put the result in input   "$.glue"   (null = discard the result)
  → OutputPath       select what passes to next state   "$"

• Keys ending in .$ are evaluated as paths or intrinsic functions; others are literal values.
• Intrinsic functions: States.Format, States.StringToJson, States.JsonToString, States.Array, States.MathAdd, States.UUID and more.
• Workflows can instead use JSONata ("QueryLanguage": "JSONata"), with Arguments and Output fields and expressions in {% … %}, plus variables that later states can read.
• State payloads are limited to 256 KB: pass S3 keys, not data.

6. Map: Fan-Out over Many Items

"LoadEachTable": {
  "Type": "Map",
  "ItemsPath": "$.tables",
  "MaxConcurrency": 5,
  "ItemSelector": { "table.$": "$$.Map.Item.Value", "run_date.$": "$.run_date" },
  "ItemProcessor": {
    "ProcessorConfig": { "Mode": "INLINE" },
    "StartAt": "LoadTable",
    "States": {
      "LoadTable": {
        "Type": "Task",
        "Resource": "arn:aws:states:::glue:startJobRun.sync",
        "Parameters": { "JobName": "generic-table-load",
                        "Arguments": { "--table.$": "$.table", "--run_date.$": "$.run_date" } },
        "End": true
      }
    }
  },
  "Next": "Done"
}

Distributed Map processes very large datasets: it reads items straight from S3 (a prefix listing, CSV, JSON lines or an S3 Inventory manifest) and runs up to thousands of child workflows in parallel:

"ProcessAllFiles": {
  "Type": "Map",
  "ItemReader": {
    "Resource": "arn:aws:states:::s3:listObjectsV2",
    "Parameters": { "Bucket": "acme-lake-prod", "Prefix": "raw/backfill/" }
  },
  "MaxConcurrency": 500,
  "ToleratedFailurePercentage": 1,
  "ItemBatcher": { "MaxItemsPerBatch": 50 },
  "ItemProcessor": {
    "ProcessorConfig": { "Mode": "DISTRIBUTED", "ExecutionType": "EXPRESS" },
    "StartAt": "ConvertBatch",
    "States": {
      "ConvertBatch": { "Type": "Task", "Resource": "arn:aws:states:::lambda:invoke",
                        "Parameters": { "FunctionName": "convert-to-parquet", "Payload.$": "$" }, "End": true }
    }
  },
  "ResultWriter": { "Resource": "arn:aws:states:::s3:putObject",
                    "Parameters": { "Bucket": "acme-lake-prod", "Prefix": "_runs/backfill/" } },
  "End": true
}

7. Error Handling

Order matters: Retry rules are tried first, and Catch runs only after retries are exhausted. Put specific errors before States.ALL.

8. Triggering, Testing & Monitoring

aws stepfunctions start-execution \
  --state-machine-arn arn:aws:states:ap-south-1:111122223333:stateMachine:daily-load \
  --name load-2026-10-26 \
  --input '{"run_date": "2026-10-26", "bucket": "acme-lake-prod", "key": "raw/txn/2026-10-26.csv.gz"}'

aws stepfunctions describe-execution --execution-arn <execution-arn>
aws stepfunctions list-executions --state-machine-arn <arn> --status-filter FAILED

• Schedules: EventBridge Scheduler (cron with time zones) targeting the state machine. Events: EventBridge rules on S3 Object Created, Glue job state changes, or custom events.
• Execution names must be unique for 90 days in Standard workflows: use them as an idempotency key (e.g. load-2026-10-26).
• Test a single state with the TestState API before deploying; use versions and aliases for safe rollouts.
• Alarm on ExecutionsFailed and ExecutionsTimedOut, and send failure events from EventBridge to SNS or chat.

9. Common Interview Questions