Historical Pipeline Instrumentation (Explicit Timestamps)

This guide covers backfilling past pipeline executions with explicit timestamps.

When This Guide Helps

Use explicit-time instrumentation when:

• You are replaying past runs into ADOC.
• You are migrating existing pipeline history.
• You need to align event and run times with the original execution time.

Historical Flow

• Create the pipeline with an explicit creation time.
• Create the run with an explicit started_at.
• Create the span with with_explicit_time=True.
• Create a job bounded by a span and fetch that span by UID.
• Create child spans under the fetched job span.
• Start, send events to, and end spans with explicit timestamps.
• Complete the run with an explicit finished_at.

Minimal Example

from datetime import datetime, timedeltafrom acceldata.client.adoc_client import AdocClientfrom acceldata.models.api.pipeline.meta import Metafrom acceldata.models.sdk.pipeline.create_job_input import CreateJobInput, JobInputOutputReffrom acceldata.models.sdk.pipeline import GenericEvent, PipelineRunResult, PipelineRunStatusfrom acceldata.models.sdk.pipeline.create_pipeline_input_request import ( CreatePipelineInputRequest,)from acceldata.models.sdk.pipeline.pipeline_run_input import PipelineRunInputclient = 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 the pipeline with a 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 the run with a 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 the span in 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 belowalternative_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 the 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 to, 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 the run with an explicit finish time.run.update_pipeline_run(    result=PipelineRunResult.SUCCESS,    status=PipelineRunStatus.COMPLETED,    finished_at=run_end,)
JobInputOutputRef(asset_uid=...) must use dot notation (<data_source_uid>.<asset_uid>). Copy the value from the Asset page in the ADOC UI.

Critical Rules

• Always set with_explicit_time=True for spans in historical mode.
• Always call start(created_at=...) explicitly for those spans.
• Always set created_at for important events.
• Always set finished_at when completing the run.

Common Mistakes

• Forgetting to call start(created_at=...) before sending events or ending spans.
• Mixing historical and current-time values in one run.
• Completing the run without finished_at.