Acceldata AWS States Adapter

acceldata_aws_states_adapter emits OpenLineage events for AWS Step Functions executions into Acceldata ADOC.

It is designed to run as an AWS Lambda (often behind EventBridge) that:

1. Receives Step Functions execution status-change events
2. Emits a START OpenLineage run when an execution begins (RUNNING)
3. Walks execution history on terminal statuses and emits OpenLineage events for individual states (Task, Map, Parallel, Choice, Wait, Pass, Succeed, Fail)
4. Sends events over HTTP to ADOC’s lineage endpoint

Python package import root: acceldata
PyPI / distribution name: acceldata_aws_states_adapter

---

Requirements

• Python >= 3.10
• Runtime dependencies: openlineage-python, boto3, attrs
• AWS permissions for the Lambda role (see IAM permissions)
• ADOC URL + credentials for the OpenLineage HTTP transport

---

Installation

From PyPI

pip install acceldata_aws_states_adapter

From TestPyPI

pip install acceldata-aws-states-adapter \
  --index-url https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/

As a Lambda layer

Build a layer zip with a python/ prefix (Lambda’s expected layout):

mkdir -p python
pip install acceldata_aws_states_adapter \
  --index-url https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/ \
  -t python/
zip -r layer.zip python

Attach layer.zip as a Lambda layer and keep your handler code (for example lambda_function.py) in the function package.

Editable / local development

cd modules/acceldata-aws-states-adapter
pip install -e .

---

How it works

EventBridge (aws.states / Step Functions Execution Status Change)
        │
        ▼
   AWS Lambda handler
        │
        ├── status == RUNNING
        │       └── ProcessorService.send_openlineage_event_for_state_machine_start(...)
        │               • describe state machine definition
        │               • build job facets (upstream / child states)
        │               • optional ParentRunFacet from execution input
        │               • emit OpenLineage START for the state machine run
        │
        └── terminal status (SUCCEEDED / FAILED / TIMED_OUT / ABORTED / ...)
                └── ProcessorService.process_step_function_execution(...)
                        • get_execution_history
                        • dispatch per-state handlers (FlowFactory)
                        • emit START/COMPLETE/FAIL/... events per state
                        • emit terminal state-machine run event + run facet

Supported Step Functions state types

State type	Handler
Task	Task
Map	Map
Parallel	Parallel
Choice	Choice
Wait	Wait
Pass	Pass
Succeed	Success
Fail	Fail

Handlers live under acceldata.step_functions.flows and are registered in FlowFactory.

OpenLineage mapping (high level)

Concept	Typical value
Job namespace	From ADOC__OPENLINEAGE__NAMESPACE, or default aws-<accountId>-<region> derived from the state machine ARN
State machine job name	State machine name (ARN suffix)
State job name	<stateMachineName>.<stateName>
State machine run id	Step Functions execution name / id
Parent run	Optional, from execution input keys aws.states.openlineage.*

Custom facets include:

• AWSStateMachineJobFacet — upstream and child state topology
• AWSStateMachineRunFacet — per-state terminal status map
• AWSStateMachineStateInputDetailsFacet / AWSStateMachineStateOutputDetailsFacet — entered/exited state details

Events are posted to ADOC at:

{ADOC__OPENLINEAGE__TRANSPORT__URL}/torch-pipeline/api/v1/lineage

using HTTP headers accessKey and secretKey.

---

Configuration

Configure the Lambda (or process) with these environment variables.

Required ADOC transport

Environment variable	Description
ADOC__OPENLINEAGE__TRANSPORT__URL	ADOC base URL (host only; endpoint path is appended by the client)
ADOC__OPENLINEAGE__TRANSPORT__ACCESS_KEY	ADOC access key
ADOC__OPENLINEAGE__TRANSPORT__SECRET_KEY	ADOC secret key (unless Secrets Manager is enabled)

Optional

Environment variable	Description
ADOC__OPENLINEAGE__NAMESPACE	OpenLineage job namespace override (recommended human-readable account/env name)
ADOC__OPENLINEAGE__TRANSPORT__SECRET_MANAGER_ENABLED	Set to true to load the secret key from AWS Secrets Manager
ADOC__OPENLINEAGE__TRANSPORT__SECRET_MANAGER_SECRET_NAME	Secrets Manager secret id/name
ADOC__OPENLINEAGE__TRANSPORT__SECRET_MANAGER_REGION	Region for the Secrets Manager client

If ADOC__OPENLINEAGE__NAMESPACE is unset, the adapter defaults to:

aws-<accountId>-<region>

parsed from the state machine ARN (for example aws-241571801844-ap-south-1).

Parent run (optional)

If the Step Functions execution input JSON includes all of:

• aws.states.openlineage.parentRunId
• aws.states.openlineage.parentJobName
• aws.states.openlineage.parentJobNamespace

the START event for the state machine run includes an OpenLineage ParentRunFacet.

Example execution input fragment:

{
  "TransactionType": "PURCHASE",
  "aws.states.openlineage.parentRunId": "1f0f2305-6017-42a6-af78-002fec15eba2",
  "aws.states.openlineage.parentJobName": "Step Functions StartExecution",
  "aws.states.openlineage.parentJobNamespace": "TriggerStateMachine"
}

---

IAM permissions

Minimum permissions typically required by the Lambda role:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "states:DescribeStateMachine",
        "states:GetExecutionHistory"
      ],
      "Resource": "*"
    }
  ]
}

If using Secrets Manager for the ADOC secret key, also allow:

{
  "Effect": "Allow",
  "Action": ["secretsmanager:GetSecretValue"],
  "Resource": "arn:aws:secretsmanager:<region>:<account-id>:secret:<secret-name>*"
}

Tighten Resource ARNs to specific state machines / secrets in production.

---

Usage (Lambda handler)

Wire EventBridge rule(s) for Step Functions Execution Status Change to a Lambda. Example handler:

import json

from acceldata.client.client_initialiser import ClientInitialiser
from acceldata.service.processor_service import ProcessorService


def lambda_handler(event, context):
    event_details = event["detail"]
    status = event_details["status"]

    client = ClientInitialiser.initialise_client_configuration().client
    service = ProcessorService(openlineage_client=client)

    if status == "RUNNING":
        service.send_openlineage_event_for_state_machine_start(
            state_machine_arn=event_details["stateMachineArn"],
            execution_id=event_details["name"],
            event_time=event_details["startDate"],
            event_details=event_details,
        )
    else:
        # SUCCEEDED / FAILED / TIMED_OUT / ABORTED / etc.
        service.process_step_function_execution(
            state_machine_arn=event_details["stateMachineArn"],
            state_machine_execution_arn=event_details["executionArn"],
            client=client,
        )

    return {"statusCode": 200, "body": json.dumps("ok")}

Expected EventBridge shape

The adapter expects the standard EventBridge envelope. The handler uses event["detail"], which includes fields such as:

• status
• name (execution id)
• stateMachineArn
• executionArn
• startDate (epoch millis)
• input (JSON string of the execution input, when included)

---

Package layout

acceldata/
  client/           # ADOC OpenLineage HTTP client + env/Secrets Manager bootstrap
  constants/        # Env var names and Step Functions field constants
  facets/           # Custom OpenLineage facets for AWS Step Functions
  service/          # ProcessorService + Step Functions describe/history services
  step_functions/   # FlowFactory, utils, per-state flow handlers

Primary entry points for integrators:

• acceldata.client.client_initialiser.ClientInitialiser
• acceldata.service.processor_service.ProcessorService

---

Building and publishing

cd modules/acceldata-aws-states-adapter
pip install -U build twine
rm -rf dist/ build/ *.egg-info
python -m build
twine check dist/*

# TestPyPI
twine upload --repository testpypi dist/*

# PyPI (after validation)
twine upload dist/*

Version is read from the VERSION file (dynamic setuptools version).

---

License

MIT — see LICENSE.