Historical Pipeline Instrumentation (Explicit Timestamps)

Use explicit timestamps to backfill past pipeline executions into Acceldata Data Observability Cloud (ADOC) with acceldata-sdk-python. In explicit-time mode, ADOC records the pipeline, run, spans, and events at the times that you specify instead of the current time.

This page assumes that you’re familiar with creating pipelines, runs, jobs, and spans, as described in Pipelines Guide. For client setup, see acceldata-sdk-python Overview.

When to use explicit timestamps

Use explicit timestamps when:

• You're replaying past pipeline runs into ADOC.
• You're migrating existing pipeline history to ADOC.
• You need run, span, and event times in ADOC to match the original execution times.

How historical instrumentation works

To backfill a historical run, create each object with an explicit timestamp in the following order:

1. Create the pipeline with an explicit creation time.
2. Create the run with an explicit started_at.
3. Create the root span with with_explicit_time=True.
4. Create a job bounded by a span, and then fetch the job span by its UID.
5. Create child spans under the fetched job span.
6. Start each span, send events to it, and end it, passing an explicit timestamp to each call.
7. Complete the run with an explicit finished_at.

Example: Backfill a historical run

The following script backfills one run of a customers ETL pipeline that ran on January 15, 2024. Replace the placeholder values for url, access_key, and secret_key with your ADOC details, and then run the script.

from datetime import datetime, timedelta
from acceldata.client.adoc_client import AdocClient
from acceldata.models.api.pipeline.meta import Meta
from acceldata.models.sdk.pipeline.create_job_input import CreateJobInput, JobInputOutputRef
from acceldata.models.sdk.pipeline import GenericEvent, PipelineRunResult, PipelineRunStatus
from acceldata.models.sdk.pipeline.create_pipeline_input_request import (
    CreatePipelineInputRequest,
)
from acceldata.models.sdk.pipeline.pipeline_run_input import PipelineRunInput
client = AdocClient(
    url="https://<your-adoc-url>",
    access_key="<your-access-key>",
    secret_key="<your-secret-key>",
)
base_time = datetime(2024, 1, 15, 10, 0, 0)
# 1) Create pipeline with historical createdAt
pipeline = client.create_pipeline(
    CreatePipelineInputRequest(
        uid="historical_customers_etl",
        name="Historical customers ETL",
        meta=Meta(owner="data-team", team="analytics", code_location="..."),
        createdAt=base_time,
    )
)
# 2) Create run with historical start time
run_start = base_time + timedelta(hours=1)
run = pipeline.create_pipeline_run(PipelineRunInput(started_at=run_start))
# `run` is a PipelineRunResource.
print(run.to_dict())
# 3) Create span with explicit-time mode (no auto-start)
root_span = run.create_root_span(
    uid="historical_customers_root",
    with_explicit_time=True,
)
# 4) Create a bounded job span and load it by UID
job = run.create_job(
    CreateJobInput(
        uid="historical_customers_transform",
        name="Historical customers transform",
        description="Backfilled transform job with explicit span timing.",
        pipeline_run_id=run.id,
        bounded_by_span=True,
        with_explicit_time=True,
        span_uid="historical_customers_transform_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={"replay": "historical_customers_etl"},
    )
)
job_span = run.get_span("historical_customers_transform_span") # explicit source for child spans below
alternative_job = run.create_job(
    CreateJobInput(
        uid="historical_customers_enrich",
        name="Historical customers enrich",
        description="Unbounded job bound to a manually created child span.",
        pipeline_run_id=run.id,
        bounded_by_span=False,
        with_explicit_time=True,
        inputs=[JobInputOutputRef(asset_uid="snowflake_ds.warehouse.customers_curated")],
        outputs=[JobInputOutputRef(asset_uid="snowflake_ds.warehouse.customers_enriched")],
        meta=Meta(owner="data-team", team="analytics", code_location="..."),
        context={"replay": "historical_customers_etl", "variant": "manual_span"},
    )
)
# Alternative with bounded_by_span=False: create and bind span explicitly.
alternative_job_span = job_span.create_child_span(
    uid="historical_customers_enrich_span",
    associated_job_uids=[alternative_job.uid],
    with_explicit_time=True,
)
# 5) Start, send events, and end spans at chosen timestamps
root_span_start = run_start + timedelta(seconds=10)
job_span_start = run_start + timedelta(minutes=1)
alternative_start = run_start + timedelta(minutes=1, seconds=30)
alternative_event = run_start + timedelta(minutes=1, seconds=45)
alternative_end = run_start + timedelta(minutes=1, seconds=55)
validate_start = run_start + timedelta(minutes=2)
validate_event = run_start + timedelta(minutes=3)
validate_end = run_start + timedelta(minutes=4)
publish_start = run_start + timedelta(minutes=5)
publish_event = run_start + timedelta(minutes=6)
publish_end = run_start + timedelta(minutes=8)
job_span_end = run_start + timedelta(minutes=9)
root_span_end = run_start + timedelta(minutes=15)
run_end = run_start + timedelta(minutes=20)
root_span.start(created_at=root_span_start)
job_span.start(created_at=job_span_start)
alternative_job_span.start(created_at=alternative_start)
alternative_job_span.send_event(
    GenericEvent(
        event_uid="historical_enrich_completed",
        context_data={"rows_enriched": 11800},
        created_at=alternative_event,
    )
)
alternative_job_span.end(created_at=alternative_end)
validate_span = job_span.create_child_span(
    uid="historical_customers_validate_span",
    associated_job_uids=[job.uid],
    with_explicit_time=True,
)
validate_span.start(created_at=validate_start)
validate_span.send_event(
    GenericEvent(
        event_uid="historical_validate_completed",
        context_data={"rows_validated": 12000},
        created_at=validate_event,
    )
)
validate_span.end(created_at=validate_end)
publish_span = job_span.create_child_span(
    uid="historical_customers_publish_span",
    associated_job_uids=[job.uid],
    with_explicit_time=True,
)
publish_span.start(created_at=publish_start)
publish_span.send_event(
    GenericEvent(
        event_uid="historical_publish_completed",
        context_data={"rows_loaded": 11800},
        created_at=publish_event,
    )
)
publish_span.end(created_at=publish_end)
job_span.end(created_at=job_span_end)
root_span.end(created_at=root_span_end)
# 5) Complete run with explicit finish time
run.update_pipeline_run(
    result=PipelineRunResult.SUCCESS,
    status=PipelineRunStatus.COMPLETED,
    finished_at=run_end,
)

Rules for explicit timestamps

• Set with_explicit_time=True on every span and bounded job in a historical run.
• Call start(created_at=...) on each span before you send events to it or end it.
• Set created_at on every event that you send.
• Set finished_at when you complete the run.

Common mistakes

• Sending events to a span or ending it without first calling start(created_at=...).
• Mixing explicit timestamps and current-time values in the same run.
• Completing the run withoutfinished_at.

What’s next

After you complete this section, explore:

• Pipelines Guide – Learn how to create pipelines, runs, jobs, and spans, including the explicit-time parameters for each.
• End-to-End Pipeline Lifecycle Example – See a complete current-time lifecycle script, including error handling.
• Pipeline Continuation IDs (Multi-Script Runs) – Learn how multiple scripts or services can update the same pipeline run.