End-to-End Pipeline Lifecycle Example

This guide walks through one script that demonstrates a complete pipeline lifecycle in ADOC using acceldata-sdk-python.

What This Example Includes

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

Full Lifecycle Script

from acceldata.client.adoc_client import AdocClientfrom acceldata.exceptions import APIError, ApiException, AcceldataSdkExceptionfrom acceldata.models.api.pipeline.meta import Metafrom acceldata.models.api.pipeline.tag import Tagfrom acceldata.models.sdk.pipeline.create_job_input import CreateJobInput, JobInputOutputReffrom acceldata.models.sdk.pipeline.create_pipeline_input_request import ( CreatePipelineInputRequest,)from acceldata.models.sdk.pipeline.events.generic_event import GenericEventfrom acceldata.models.sdk.pipeline.events.log_event import LogEventfrom 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.SUCCESSstatus = PipelineRunStatus.COMPLETEDroot_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__})    raiseexcept Exception as err:    result = PipelineRunResult.FAILURE    status = PipelineRunStatus.FAILED    root_span.failed(context_data={"error": str(err), "type": "UnhandledException"})    raisefinally:    run.update_pipeline_run(        result=result,        status=status,        context_data={"example": "e2e-lifecycle"},    )    print(run.to_dict())

JobInputOutputRef(asset_uid=...) must use dot notation (<data_source_uid>.<asset_uid>). Copy the value from the Asset page in the ADOC UI.

Lifecycle Checklist

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.