End-to-End Pipeline Lifecycle Example

This example uses one Python script to run a complete pipeline lifecycle in ADOC with acceldata-sdk-python.

What the script does

The script performs the following steps in order:

1. Create a client and a pipeline.
2. Start a run.
3. Create a root span and a job bounded by a span.
4. Fetch a job span by UID and create child spans under it.
5. Send LogEvent and GenericEvent.
6. End child spans, the bounded job span, and the root span.
7. Mark the run COMPLETED on success or FAILED on error.

Run the lifecycle script

Replace the placeholder values for url, access_key, and secret_key with your ADOC details, and then run the script.

from acceldata.client.adoc_client import AdocClient
from acceldata.exceptions import APIError, ApiException, AcceldataSdkException
from acceldata.models.api.pipeline.meta import Meta
from acceldata.models.api.pipeline.tag import Tag
from acceldata.models.sdk.pipeline.create_job_input import CreateJobInput, JobInputOutputRef
from acceldata.models.sdk.pipeline.create_pipeline_input_request import (
    CreatePipelineInputRequest,
)
from acceldata.models.sdk.pipeline.events.generic_event import GenericEvent
from acceldata.models.sdk.pipeline.events.log_event import LogEvent
from acceldata.models.sdk.pipeline.pipeline_run_input import (
    PipelineRunResult,
    PipelineRunStatus,
)
client = AdocClient(
    url="https://<your-adoc-url>",
    access_key="<your-access-key>",
    secret_key="<your-secret-key>",
)
pipeline = client.create_pipeline(
    CreatePipelineInputRequest(
        uid="e2e_customers_pipeline",
        name="E2E customers pipeline",
        description="End-to-end pipeline lifecycle example",
        meta=Meta(owner="data-team", team="analytics", code_location="..."),
        tags=[
            Tag(name="env:prod", displayName="Environment: Production"),
            Tag(name="domain:finance", displayName="Domain: Finance"),
        ],
    )
)
run = pipeline.create_pipeline_run()
result = PipelineRunResult.SUCCESS
status = PipelineRunStatus.COMPLETED
root_span = run.create_root_span(uid="e2e_customers_root")
try:
    root_span.send_event(LogEvent("Run started", context_data={"run_id": run.id}))
    transform_job = run.create_job(
        CreateJobInput(
            uid="customers_transform",
            name="Transform customers",
            description="End-to-end example: land curated customers for downstream analytics.",
            pipeline_run_id=run.id,
            bounded_by_span=True,
            with_explicit_time=False,
            span_uid="customers_transform_job_span",
            inputs=[JobInputOutputRef(asset_uid="s3_ds.staging.customers_raw_snapshot")],
            outputs=[JobInputOutputRef(asset_uid="snowflake_ds.warehouse.customers_curated")],
            meta=Meta(owner="data-team", team="analytics", code_location="..."),
            context={"example": "e2e_customers_pipeline"},
        )
    )
    # Load the bounded span by UID so the example has an explicit `job_span` source.
    transform_job_span = run.get_span("customers_transform_job_span")
    transform_job_span.send_event(
        GenericEvent(
            event_uid="transform_job_started",
            context_data={"job_uid": transform_job.uid},
        )
    )
    validate_span = transform_job_span.create_child_span(
        uid="customers_transform_validate_span",
        associated_job_uids=[transform_job.uid],
    )
    validate_span.send_event(
        GenericEvent(
            event_uid="validate_phase_started",
            context_data={"checks": "schema_and_nulls"},
        )
    )
    validate_span.send_event(
        LogEvent("Validation completed", context_data={"failed_checks": 0})
    )
    validate_span.end()
    publish_span = transform_job_span.create_child_span(
        uid="customers_transform_publish_span",
        associated_job_uids=[transform_job.uid],
    )
    publish_span.send_event(
        GenericEvent(
            event_uid="publish_phase_started",
            context_data={"target": "warehouse"},
        )
    )
    publish_span.send_event(
        LogEvent(
            "Publish completed",
            context_data={"rows_before": 120000, "rows_after": 118742},
        )
    )
    publish_span.end()
    transform_job_span.end()
    root_span.send_event(
        GenericEvent(
            event_uid="pipeline_business_metrics",
            context_data={"rows_loaded": 118742},
        )
    )
    root_span.end(context_data={"dag_status": "SUCCESS"})
except (APIError, ApiException, AcceldataSdkException) as err:
    result = PipelineRunResult.FAILURE
    status = PipelineRunStatus.FAILED
    root_span.failed(context_data={"error": str(err), "type": type(err).__name__})
    raise
except Exception as err:
    result = PipelineRunResult.FAILURE
    status = PipelineRunStatus.FAILED
    root_span.failed(context_data={"error": str(err), "type": "UnhandledException"})
    raise
finally:
    run.update_pipeline_run(
        result=result,
        status=status,
        context_data={"example": "e2e-lifecycle"},
    )
    print(run.to_dict())

Configure the job

The script creates the job with CreateJobInput. The following parameters define the job, its bounded span, and its input and output assets:

Create child spans

The script calls create_child_span() on the job span to track the validate and publish phases. The method uses the following parameters:

Set the run outcome

The script sets a successful outcome before the try block and changes it if an exception occurs. The finally block always calls update_pipeline_run(), so the run always reaches a terminal state:

Check the lifecycle order

Before you adapt this script, confirm each of the following:

• The run is created before any spans, jobs, or events.
• The job uses bounded_by_span=True with a span_uid.
• The job span is fetched by UID before you create child spans under it.
• Child spans are ended before you end the bounded job span.
• The bounded job span is ended before you end the root span.
• The run is marked terminal (COMPLETED or FAILED) in all cases.

What's next

After you complete this section, explore:

• Pipeline Continuation IDs (Multi-Script Runs): Learn how multiple scripts or services can update the same pipeline run.
• Historical Pipeline Instrumentation (Explicit Timestamps): Learn how to backfill past pipeline executions with explicit timestamps.