Pipelines Guide

This guide describes how to record pipeline executions in ADOC with

acceldata-sdk-python.

Introduction

ADOC treats data pipelines as first-class observability objects. Each definition captures how work is structured (assets and jobs), and each run captures what happened during a given execution. acceldata-sdk-python creates those definitions, opens runs, attaches hierarchical spans, emits metrics and custom events, and closes runs with a clear outcome, so the Catalog and Pipelines views in ADOC stay aligned with your orchestration code.

Core Concepts

Pipeline APIs focus on orchestration observability: definitions, runs, spans, jobs, and events. Catalog APIs focus on what is stored and how it is profiled or crawled. Most production setups use both: you register and profile assets through catalog calls, while pipeline calls describe how those assets are produced or consumed in each batch or stream.

The rest of this guide is organized by topic: definition, runs, spans, jobs, events, completion, and retrieval. A full execution typically defines or loads a pipeline, opens a run, records spans and jobs, emits events as work proceeds, and then updates the run to a terminal state.

Creating a Pipeline

Use CreatePipelineInputRequestto define and create a pipeline in ADOC. This request includes required identifiers along with several optional fields for metadata and organization.

Key Fields

Tags (Optional)

Recommended Approach

Use tag_associations with tag_write_mode="ASSOCIATIONS" for all new implementations. Tag associations reference existing Tag Service keys and values using their IDs (keyId and keyValueId).

The legacy tags field will continue to appear in API responses for backward compatibility.

For instructions on creating tag keys and values and retrieving their IDs, see the Tags guide.

Starting with acceldata-sdk-python 26.9.0, you can use tag_write_mode to specify how tags are applied when creating or updating pipelines.

This option is available in the following SDK interfaces:

• CreatePipelineInputRequest
• AdocClient.create_pipeline()
• AdocClient.update_pipeline()

The tag_write_mode parameter determines which tag payload the SDK applies to the pipeline. When you update a pipeline, the provided tags replace all existing tag associations. Passing an empty list removes all tags from the pipeline.

Create a Pipeline Using Tag Associations (Recommended)

Use tag_associations with tag_write_mode="ASSOCIATIONS" to associate existing Tag Service keys and values with a pipeline.

Before you begin, ensure that the required tag keys and values exist in the Tag Service. For instructions on creating tags and retrieving their IDs, see the Unified Tags Guide.

Example:

from acceldata.models.api.pipeline.tag_detail import TagDetail
from acceldata.models.sdk.pipeline.create_pipeline_input_request import (
    CreatePipelineInputRequest,
)
pipeline = client.create_pipeline(
    CreatePipelineInputRequest(
        uid="monthly_reporting_pipeline_tagged",
        name="Monthly reporting pipeline (associations)",
        tag_associations=[
            TagDetail(keyId=12, keyValueId=34),
            TagDetail(keyId=12, keyValueId=56),
        ],
        tag_write_mode="ASSOCIATIONS",
    )
)

In this example, the SDK associates two existing Tag Service key-value pairs with the new pipeline.

Create a Pipeline Using Legacy Tags (Deprecated)

The TAGS mode supports the legacy tagging format. Each tag contains the following fields:

• name: A stable tag identifier, typically formatted as a key-value pair (for example, env:prod).
• displayName: A human-readable tag label displayed in the UI.

When you create a pipeline using this mode, the SDK uses displayName to find or create the corresponding Tag Service key and value.

Example:

from acceldata.models.api.pipeline.tag import Tag
from acceldata.models.sdk.pipeline.create_pipeline_input_request import (
    CreatePipelineInputRequest,
)
# tag_write_mode="TAGS" is the default; shown here for clarity
pipeline = client.create_pipeline(
    CreatePipelineInputRequest(
        uid="monthly_reporting_pipeline",
        name="Monthly reporting pipeline",
        tags=[
            Tag(name="env:prod", displayName="Environment: Production"),
            Tag(name="domain:finance", displayName="Domain: Finance"),
        ],
        tag_write_mode="TAGS",
    )
)

Replace Tags on an Existing Pipeline

You can replace the tags associated with an existing pipeline using PipelineResource.replace_tags() or AdocClient.replace_pipeline_tags().

Both methods replace all existing tag associations with the supplied list. To add or remove individual tags, provide the complete list of tags that you want to retain.

Replace Tags Using Tag Associations (Recommended)

Use tag_associations with tag_write_mode="ASSOCIATIONS" to replace pipeline tags using existing Tag Service key and value IDs.

You can replace tags using either of the following methods.

Option 1: Using PipelineResource.replace_tags()

from acceldata.models.api.pipeline.tag_detail import TagDetail
pipeline.replace_tags(
    tag_associations=[
        TagDetail(keyId=12, keyValueId=34),
        TagDetail(keyId=15, keyValueId=78),
    ],
    tag_write_mode="ASSOCIATIONS",
)

Option 2: Using AdocClient.replace_pipeline_tags()
client.replace_pipeline_tags(
    pipeline_uid,
    tag_associations=[TagDetail(keyId=12, keyValueId=34)],
    tag_write_mode="ASSOCIATIONS",
)

Both methods replace the pipeline's existing tag associations with the specified key-value pairs.

Replace Tags Using Legacy Tags (Deprecated)

The legacy replace_tags() and replace_pipeline_tags() methods support display-name-based tags through the default tag_write_mode="TAGS".

Deprecation Notice: Starting with acceldata-sdk-python 26.9.0, writing display-name-based tags using these methods is deprecated and scheduled for removal after three releases.

For new implementations, use tag_associations with tag_write_mode="ASSOCIATIONS".

Example:

from acceldata.models.api.pipeline.tag import Tag
pipeline.replace_tags(
    [
        Tag(name="env:stage", displayName="Environment: Staging"),
        Tag(name="owner:data-platform", displayName="Owner: Data Platform"),
    ]
)

This example replaces all existing pipeline tags with the specified legacy tags.

Manage Pipeline Runs and Spans

You can use the Python SDK to create pipeline runs, track execution progress, and record events using spans.

A pipeline run represents a single execution of a pipeline. Spans represent individual operations within that execution and help you monitor execution progress, capture metadata, and track the status of each operation.

Create a Pipeline Run

Use PipelineResource.create_pipeline_run() to start a pipeline run.

run_resource = pipeline_resource.create_pipeline_run()
LOGGER.info("Started pipeline run id=%s", run_resource.id)

The method returns a PipelineRunResource object that you can use to manage the run.

The PipelineRunResource provides the following methods:

Create a Root Span

A root span represents the top-level operation within a pipeline run. You can create child spans under the root span to track individual tasks or processing steps.

Use create_root_span() to create a root span. By default, the SDK initializes the span with the current timestamp.

You can also attach metadata to the root span using events.

root_span_resource = run_resource.create_root_span(uid="snippet_customers_root")
root_span_resource.send_event(LogEvent("Run started", context_data={"run_id": run_resource.id}))

In this example, the SDK creates a root span and records a log event containing the pipeline run ID.

Create Child Spans

Use SpanResource.create_child_span() to create child spans under a root span or another span.

Child spans help you track individual operations within a pipeline run, such as data extraction, transformation, or loading.

The method supports the following parameters:

Manage Span Status and Events

The SpanResource provides methods to inspect the span hierarchy and update execution status.

You can also record execution information using the following event types:

• LogEvent: Records log messages and related context.
• GenericEvent: Records custom events, such as processing metrics and execution results.

Example: Track a Pipeline Run with Child Spans

The following example demonstrates how to create a pipeline run, track an operation using a child span, record processing metrics, and mark the run as completed.

from datetime import datetime
from acceldata.models.sdk.pipeline.events.generic_event import GenericEvent
from acceldata.models.sdk.pipeline.pipeline_run_input import (
    PipelineRunResult,
    PipelineRunStatus,
)
pipeline_run = pipeline_resource.create_pipeline_run()
# ``get_root_span()`` returns the run root when present; 
root_span = run_resource.get_root_span()
# is_root returns true if the span is root else false; 
is_root = bool(root_span.is_root())
# is_root returns true if the span has children else false;
has_children = bool(root_span.has_children())
span_ctx = run_resource.create_root_span(uid="monthly.generate.data.span")
span_ctx.start()
child_span = span_ctx.create_child_span("monthly.generate.customer.span")
child_span.send_event(
    GenericEvent(
        event_uid="order.customer.join.result",
        context_data={"client_time": str(datetime.now()), "row_count": 100},
    )
)
child_span.end()
span_ctx.end()
run_resource.update_pipeline_run(
    context_data={"name": "backend", "key1": "value2"},
    result=PipelineRunResult.SUCCESS,
    status=PipelineRunStatus.COMPLETED,
)

Create and Manage Pipeline Jobs

You can use the Python SDK to create jobs within a pipeline run, associate jobs with input and output assets, and track job execution using spans.

A job represents an individual processing operation within a pipeline, such as extracting data, transforming records, or generating reports. You can associate a job with a span to monitor its execution and record events.

Create a Pipeline Job

You can create a job using either PipelineResource.create_job() or PipelineRunResource.create_job().

Both methods create a job within a pipeline run. The difference is how you specify the pipeline run ID.

Configure Job Parameters

Use CreateJobInput to define the job and its associated metadata.

Create a Job Using a Pipeline Resource

Use PipelineResource.create_job() when you have a pipeline resource and an existing pipeline run ID.

from acceldata.models.api.pipeline.meta import Meta
from acceldata.models.sdk.pipeline.create_job_input import CreateJobInput, JobInputOutputRef
node_from_pipeline = pipeline_resource.create_job(
    CreateJobInput(
        uid="monthly_sales_aggregate",
        name="Monthly sales aggregate",
        description="Builds monthly aggregates for reporting.",
        pipeline_run_id=run_resource.id,
        inputs=[JobInputOutputRef(asset_uid="warehouse_ds.db.schema.source_table")],
        outputs=[JobInputOutputRef(job_uid="downstream_job_uid")],
        meta=Meta(owner="analytics", team="finance", code_location="..."),
        context={"batch": "2026-04"},
        bounded_by_span=True,
        span_uid="monthly_sales_aggregate_span",
    )
)

This example creates a job associated with the specified pipeline run. The SDK also creates a span to track the job's execution.

Create a Job Using a Pipeline Run Resource

Use PipelineRunResource.create_job() when you already have a pipeline run resource.

You don't need to specify pipeline_run_id because the SDK automatically associates the job with the current run.

from acceldata.models.api.pipeline.meta import Meta
from acceldata.models.sdk.pipeline.create_job_input import CreateJobInput, JobInputOutputRef
node_from_run = run_resource.create_job(
    CreateJobInput(
        uid="monthly_sales_aggregate_r",
        name="Monthly sales aggregate",
        description="Same job with a shorter call site when the run object is already available.",
        inputs=[JobInputOutputRef(asset_uid="warehouse_ds.db.schema.source_table")],
        outputs=[JobInputOutputRef(job_uid="downstream_job_uid")],
        meta=Meta(owner="analytics", team="finance", code_location="..."),
        context={"batch": "2026-04"},
        bounded_by_span=True,
        span_uid="monthly_sales_aggregate_r_span",
    )
)

Associate Jobs with Spans

You can associate a job with a span to monitor its execution and record events.

The SDK supports two approaches:

• Automatic span creation: Set bounded_by_span=True to create a span when creating the job.
• Manual span creation: Set bounded_by_span=False and create a child span associated with the job.

Create a Job with an Associated Span (Recommended)

Set bounded_by_span=True and provide a unique span_uid to create a job and its associated span in one operation.

If a root span doesn't exist, the SDK creates one automatically.

from acceldata.models.api.pipeline.meta import Meta
from acceldata.models.sdk.pipeline.create_job_input import (
    CreateJobInput,
    JobInputOutputRef,
)
from acceldata.models.sdk.pipeline.events.generic_event import (
    GenericEvent,
)
job_node = run_resource.create_job(
    CreateJobInput(
        uid="build_reports",
        name="Build reports",
        description="Materializes monthly reporting tables.",
        pipeline_run_id=run_resource.id,
        bounded_by_span=True,
        with_explicit_time=False,
        span_uid="build_reports_span",
        inputs=[
            JobInputOutputRef(
                asset_uid="snowflake_ds.raw.input_table"
            )
        ],
        outputs=[
            JobInputOutputRef(
                asset_uid="snowflake_ds.curated.output_table"
            )
        ],
        meta=Meta(
            owner="data-team",
            team="analytics",
            code_location="..."
        ),
        context={
            "batch_id": "2026-04",
            "downstream": "looker",
        },
    )
)
job_span = run_resource.get_span("build_reports_span")
job_span.send_event(
    GenericEvent(
        event_uid="build_reports_started",
        context_data={"job_uid": job_node.uid},
    )
)

The SDK creates the job and associated span. The span starts automatically when created.

Create a Job with a Manually Associated Span

Set bounded_by_span=False to create a job without automatically creating its span. Then create a child span and associate it with the job using associated_job_uids.

from acceldata.models.api.pipeline.meta import Meta
from acceldata.models.sdk.pipeline.create_job_input import (
    CreateJobInput,
    JobInputOutputRef,
)
from acceldata.models.sdk.pipeline.events.generic_event import (
    GenericEvent,
)
manual_job = run_resource.create_job(
    CreateJobInput(
        uid="build_reports_manual",
        name="Build reports manual span binding",
        description="Creates a job with a manually associated span.",
        pipeline_run_id=run_resource.id,
        bounded_by_span=False,
        with_explicit_time=False,
        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={"bind_mode": "manual_span"},
    )
)
root_span = run_resource.get_root_span()
manual_job_span = root_span.create_child_span(
    uid="build_reports_manual_span",
    associated_job_uids=[manual_job.uid],
)
manual_job_span.send_event(
    GenericEvent(
        event_uid="manual_job_started",
        context_data={"job_uid": manual_job.uid},
    )
)
manual_job_span.end()

Reference Input and Output Assets

Use JobInputOutputRef to associate a job with input and output assets or other jobs.

Specify an asset UID using asset_uid. You can copy the asset UID from the ADOC UI.

from acceldata.models.sdk.pipeline.create_job_input import (
    JobInputOutputRef,
)
inputs = [
    JobInputOutputRef(
        asset_uid="snowflake_ds.raw.input_table"
    )
]
outputs = [
    JobInputOutputRef(
        asset_uid="snowflake_ds.curated.output_table"
    )
]

To reference another job, use job_uid instead of asset_uid.

Track Job Execution with Child Spans

Create child spans under a job span to track individual processing steps, such as data extraction, transformation, and publishing.

from acceldata.models.sdk.pipeline.events.generic_event import (
    GenericEvent,
)
job_span = run_resource.get_span("build_reports_span")
extract_span = job_span.create_child_span(
    uid="build_reports_extract_span",
    associated_job_uids=[job_node.uid],
)
extract_span.send_event(
    GenericEvent(
        event_uid="extract_phase_started",
        context_data={"step": "source_extract"},
    )
)
extract_span.end()
publish_span = job_span.create_child_span(
    uid="build_reports_publish_span",
    associated_job_uids=[job_node.uid],
)
publish_span.send_event(
    GenericEvent(
        event_uid="publish_phase_started",
        context_data={"step": "warehouse_publish"},
    )
)
publish_span.end()
job_span.end()

Each child span represents a processing phase and records an execution event. After the processing steps finish, the SDK closes the child spans and then the parent job span.

Record Span Events

Use span events to record execution logs, processing metrics, and checkpoints.

Send Log and Generic Events

The following example records a log message when report generation starts and a custom event containing the number of tables created.

from acceldata.models.sdk.pipeline.events.generic_event import (
    GenericEvent,
)
root_span_resource.send_event(
    LogEvent(
        "Started report generation",
        context_data={"batch": "2026-04"},
    )
)
root_span_resource.send_event(
    GenericEvent(
        event_uid="reports_generated",
        context_data={"tables_created": 4},
    )
)

Complete a Span

Use end() to mark a span as completed when its operation finishes successfully.

root_span_resource.end(
    context_data={"dag_status": "SUCCESS"}
)

For unsuccessful operations, use one of the following methods instead:

• failed(): Marks the span as failed.
• abort(): Marks the span as aborted.

Close all open child spans before completing their parent span.

Complete a Pipeline Run

Use PipelineRunResource.update_pipeline_run() to update a pipeline run's execution status and result.

from acceldata.models.sdk.pipeline.pipeline_run_input import (
    PipelineRunResult,
    PipelineRunStatus,
)
run_resource.update_pipeline_run(
    result=PipelineRunResult.SUCCESS,
    status=PipelineRunStatus.COMPLETED,
)

The method returns the same PipelineRunResource with an updated payload.

Pipeline Run Status and Result

Use the following combinations to represent the pipeline run's state:

Retrieve and Manage Pipelines

Use the Python SDK to retrieve existing pipelines, inspect their execution history, and manage pipeline resources.

Retrieve a Pipeline

Use AdocClient.get_pipeline() to retrieve a pipeline by its UID or numeric ID represented as a string.
pipeline_resource = client.get_pipeline(
    "monthly_reporting_pipeline"
)

The returned PipelineResource provides methods for creating pipeline runs, retrieving execution history, and creating jobs.

Retrieve the Latest Pipeline Run

Use PipelineResource.get_latest_pipeline_run() to retrieve the latest run associated with a pipeline.

pipeline_resource = client.get_pipeline(
    "monthly_reporting_pipeline"
)

pipeline_run = pipeline_resource.get_latest_pipeline_run()

Retrieve a Specific Pipeline Run

Use AdocClient.get_pipeline_run() to retrieve a pipeline run using its run ID or continuation ID.

Example: Using a Pipeline Run ID

pipeline_run = client.get_pipeline_run(
    pipeline_run_id=run_resource.id
)

Example: Using a Continuation ID

pipeline_run = client.get_pipeline_run(
    continuation_id="demo_do_not_end_pipeline_run_8",
    pipeline_id=pipeline_resource.id,
)

Alternatively, retrieve the run directly from the pipeline resource.

pipeline_run = pipeline_resource.get_run(
    continuation_id="demo_do_not_end_pipeline_run_8"
)

Retrieve Pipeline Details

Use PipelineRunResource.get_details() to retrieve pipeline details associated with a run.

pipeline_details = run_resource.get_details()

Retrieve All Spans for a Run

Use PipelineRunResource.get_spans() to retrieve the spans associated with a pipeline run.

pipeline_run_spans = run_resource.get_spans()

Retrieve All Runs for a Pipeline

Use AdocClient.get_pipeline_runs() or PipelineResource.get_runs() to retrieve the execution history of a pipeline.

runs = client.get_pipeline_runs(
    pipeline_resource.id
)

runs = pipeline_resource.get_runs()

List and Filter Pipelines

Use AdocClient.get_pipelines() to retrieve a paginated list of pipeline summaries.

The method calls GET /pipelines/summary and returns pipeline metadata and the latest execution information, when available.

Query Parameters

Supported Pipeline Filters

The SDK converts the filter fields to their corresponding API query parameters using camelCase naming.

List All Pipelines

Use get_pipelines() without filters to retrieve the first page of pipeline summaries.

listing = client.get_pipelines()
for row in listing.pipelines:
    print(
        row.pipeline_summary.uid,
        row.pipeline_summary.name,
    )

By default, the SDK returns the first 25 results.

Filter and Sort Pipelines

Use PipelineSummaryFilter to retrieve pipelines based on name, owner, tags, execution result, or other attributes.

from acceldata.models.sdk.pipeline.pipeline_summary_filter import (
    PipelineSummaryFilter,
)
listing = client.get_pipelines(
    pipeline_filter=PipelineSummaryFilter(
        name="reporting",
        owner="data-team",
        tag_key_value_ids="34,56",
        run_result="SUCCESS",
    ),
    page=0,
    size=10,
    sort_by="name:asc",
)

This example retrieves up to 10 pipelines matching the specified criteria and sorts the results by name in ascending order.

Filter Pipelines by ID

You can provide pipeline IDs as a list of integers or a comma-separated string.

from acceldata.models.sdk.pipeline.pipeline_summary_filter import (
    PipelineSummaryFilter,
)
listing = client.get_pipelines(
    pipeline_filter=PipelineSummaryFilter(
        pipeline_ids=[487648, 489112],
    ),
)

Filter Pipelines by Tags

Use Tag Service key and value IDs to retrieve pipelines associated with specific tags.

from acceldata.models.sdk.pipeline.pipeline_summary_filter import (
    PipelineSummaryFilter,
)
listing = client.get_pipelines(
    pipeline_filter=PipelineSummaryFilter(
        tag_key_ids=[12],
        tags=[34, 56],
    ),
)

The tags filter is an alias for tag_key_value_ids.

Access Pipeline Summary Details

Use the returned listing to access pipeline metadata and the latest execution result.

for row in listing.pipelines:
    summary = row.pipeline_summary
    print(
        summary.uid,
        summary.name,
        summary.owner,
        row.latest_run_result,
    )

Delete a Pipeline

Use PipelineResource.delete() to delete an existing pipeline.

delete_response = pipeline_resource.delete()

[Important]: Verify that you've selected the correct pipeline before calling delete().

Best Practices for Managing Jobs and Spans

• Create a pipeline run before creating jobs. Each job must belong to an existing run.
• Use unique identifiers. Ensure that job and span UIDs are unique within the pipeline run.
• Record events while spans are active. Don't send events after a span has ended.
• Close spans after execution. Use end() for successful operations and failed() or abort() for unsuccessful operations.
• Update the pipeline run status. Set the appropriate terminal status and result when execution finishes so ADOC accurately reflects the outcome.

What's next

After you complete this section, explore:

• End-to-End Pipeline Lifecycle Example: See a complete script that creates a pipeline, runs it, and closes it out.
• 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.
• Data Sources and Assets Guide: Learn how job inputs and outputs reference catalog assets.