acceldata-aws-states-adapter

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

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

OpenLineage mapping (high level)

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

Optional

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.