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, 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 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 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 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.

Last updated on

Was this page helpful?