Create and Execute Your First Pipeline

This is your hands-on guide to creating a complete data pipeline from scratch in ADOC. You'll learn the full lifecycle: defining the pipeline structure, setting up jobs, creating execution tracking, and monitoring the results.

Why This Matters

Creating a pipeline in ADOC isn't just about moving data - it's about establishing observability. Every step you define here becomes visible, traceable, and debuggable. When something goes wrong (and it will), you'll know exactly where, when, and why.

Real-World Scenario

The Challenge

Your company needs to sync customer data daily from your Athena data lake to Redshift for analytics. The business team needs:

• Fresh data every morning by 8 AM

• Quality checks to catch issues early

• Clear visibility when something breaks

• Ability to trace data lineage

The Solution

Build a fully observable pipeline with three jobs:

1. Extract: Pull customer data from Athena

2. Transform: Clean and validate the data

3. Load: Write to Redshift for analytics

Each step will be tracked with spans and events, giving you complete visibility into execution.

What You'll Build

┌─────────────────────────────────────────────────────────────┐
│                    Customer ETL Pipeline                      │
├─────────────────────────────────────────────────────────────┤
│                                                               │
│  [Athena Table]                                              │
│        │                                                      │
│        ├──→ [Extract Job] ──→ Span: span-extract            │
│                   │                                          │
│                   ├──→ [Transform Job] ──→ Span: span-transform │
│                            │                                 │
│                            ├──→ [Load Job] ──→ Span: span-load  │
│                                     │                        │
│                                     └──→ [Redshift Table]    │
│                                                               │
└─────────────────────────────────────────────────────────────┘

Prerequisites

• API Credentials: accessKey and secretKey

• Data Source UID: Your Athena table identifier (e.g., AwsDataCatalog.production.customers)

• Data Destination UID: Your Redshift table identifier (e.g., warehouse.public.customers)

• Understanding: Basic ETL concepts

Tip

The Complete Workflow

We'll execute 8 steps using 6 APIs:

1. Design your pipeline (planning)

2. Create the pipeline structure

3. Create a pipeline run

4. Define job nodes (3 jobs)

5. Create spans for tracking (4 spans)

6. Start execution

7. Record events as work progresses

8. Mark completion

Step 1: Design Your Pipeline

Before touching any APIs, map out your pipeline on paper.

Questions to Answer

1. What data are you moving?

   • Source: Athena table AwsDataCatalog.production.customers

   • Destination: Redshift table warehouse.public.customers

2. What transformations are needed?

   • Remove duplicate customer IDs

   • Validate email formats

   • Calculate customer lifetime value

3. What are the dependencies?

   • Extract must complete before Transform

   • Transform must complete before Load

4. Who owns this?

   • Team: data-engineering

   • Owner: data-team@company.com

   • On-call: #data-alerts Slack channel

Pipeline Design

Pipeline: "Customer ETL Pipeline"
UID: "customer-etl-daily"
Schedule: Manual (for now, we'll automate later)

Jobs:
  1. Extract (uid: job-extract-customers)
     - Input: Athena table
     - Output: None (passes data in memory)
  
  2. Transform (uid: job-transform-customers)
     - Input: Extract job output
     - Output: None (passes data in memory)
  
  3. Load (uid: job-load-redshift)
     - Input: Transform job output
     - Output: Redshift table

Checkpoint: You should have job names, UIDs, and data sources documented.

Step 2: Create the Pipeline

Register your pipeline in ADOC.

API Call

PUT /torch-pipeline/api/pipelines

Parameters: None

Request

{
 "pipeline": {
 "name": "Customer ETL Pipeline",
 "description": "Daily customer data sync from Athena to Redshift",
 "uid": "customer-etl-daily",
 "enabled": true,
 "scheduled": false,
 "schedulerType": "INTERNAL",
 "tags": ["production", "daily", "customer-data"],
 "meta": {
 "owner": "data-team@company.com",
 "team": "data-engineering",
 "codeLocation": "https://github.com/company/pipelines/customer-etl",
 "slackChannel": "#data-alerts",
 "sla": "30 minutes"
 }
 }
}

Field Explanations

Success Response

{
 "pipeline": {
 "id": 15,
 "uid": "customer-etl-daily",
 "name": "Customer ETL Pipeline",
 "enabled": true,
 "createdAt": "2024-12-05T10:00:00Z"
 }
}

Save This: pipeline.id = 15 - You'll need this for the next steps!

Step 3: Create a Pipeline Run

A "run" is a single execution instance of your pipeline. Think of it like pressing "play" - you're about to execute all the jobs.

API Call

POST /torch-pipeline/api/pipelines/15/runs

Path Parameters:

Request

{
 "run": {
 "continuationId": "run-2024-12-05-001"
 }
}

What is continuationId? A unique identifier for this specific run. Use format: run-YYYY-MM-DD-NNN where NNN is a sequence number.

Success Response

{
 "run": {
 "id": 109133,
 "pipelineId": 15,
 "continuationId": "run-2024-12-05-001",
 "status": "CREATED"
 }
}

Save This: run.id = 109133 - You'll use this for jobs and spans!

Step 4: Define Job Nodes

Jobs are the actual work units in your pipeline. You'll create three jobs that form the ETL chain.

API Call

PUT /torch-pipeline/api/pipelines/15/jobs

Path Parameters:

Important: Call this endpoint three times (once per job).

Job 1: Extract Customer Data

This job reads from your Athena table.

{
 "name": "Extract Customer Data",
 "uid": "job-extract-customers",
 "pipeLineRunId": 109133,
 "inputs": [],
 "outputs": [
 {
 "source": "ATHENA-DS",
 "asset_uid": "AwsDataCatalog.production.customers"
 }
 ],
 "meta": {
 "owner": "data-team@company.com",
 "team": "data-engineering",
 "estimatedDuration": "5 minutes",
 "dataVolume": "~100K rows"
 }
}

Why no inputs? This is the first job - it starts from a data source, not from another job.

What's asset_uid? The fully qualified name of your Athena table in ADOC's asset catalog.

Job 2: Transform Customer Data

This job processes the data from the Extract job.

{
 "name": "Transform Customer Data",
 "uid": "job-transform-customers",
 "pipeLineRunId": 109133,
 "inputs": [
 {
 "jobUid": "job-extract-customers"
 }
 ],
 "outputs": [],
 "meta": {
 "owner": "data-team@company.com",
 "team": "data-engineering",
 "transformations": ["deduplicate", "validate_emails", "calculate_ltv"]
 }
}

Key Point: inputs references the Extract job by its jobUid. This creates the dependency chain.

Job 3: Load to Redshift

This job writes the transformed data to Redshift.

{
 "name": "Load to Redshift",
 "uid": "job-load-redshift",
 "pipeLineRunId": 109133,
 "inputs": [
 {
 "jobUid": "job-transform-customers"
 }
 ],
 "outputs": [
 {
 "source": "REDSHIFT-DS",
 "asset_uid": "warehouse.public.customers"
 }
 ],
 "meta": {
 "owner": "data-team@company.com",
 "team": "data-engineering",
 "loadMethod": "UPSERT"
 }
}

The Flow: Extract → Transform → Load

Checkpoint: You should have created 3 jobs. They define WHAT to do, but haven't executed yet.

Step 5: Create Spans to Track Execution

Spans are how ADOC tracks execution. Each span represents a unit of work being performed.

API Call

POST /torch-pipeline/api/pipelines/runs/109133/spans

Path Parameters:

Important: Call this endpoint four times (1 root + 3 job spans).

Span 1: Root Span (Pipeline Level)

This represents the entire pipeline execution.

{
 "span": {
 "uid": "span-pipeline-root"
 }
}

Response:

{
 "span": {
 "id": 5000,
 "uid": "span-pipeline-root"
 }
}

Save This: span.id = 5000 - You'll use this as parentSpanId for job spans!

Span 2: Extract Span

{
 "span": {
 "uid": "span-extract",
 "parentSpanId": 5000
 }
}

Span 3: Transform Span

{
 "span": {
 "uid": "span-transform",
 "parentSpanId": 5000
 }
}

Span 4: Load Span

{
 "span": {
 "uid": "span-load",
 "parentSpanId": 5000
 }
}

Understanding Span Hierarchy:

span-pipeline-root (5000)
  ├── span-extract (5001)
  ├── span-transform (5002)
  └── span-load (5003)

Checkpoint: You've created the execution tracking structure. Now it's time to actually run!

Step 6: Start Execution

Mark the run as started - this signals that work is beginning.

API Call

PUT /torch-pipeline/api/pipelines/runs/109133

Path Parameters:

Request

{
 "run": {
 "status": "STARTED"
 }
}

Status Change: CREATED → STARTED

Step 7: Record Span Events

As each job executes, record events to track progress. This is what makes your pipeline observable!

API Call

POST /torch-pipeline/api/pipelines/spans/<span-id>/events

Path Parameters:

Event Flow for Each Span

For each of your 3 job spans (extract, transform, load), record these events:

1. Start Event (When job begins)

{
 "eventType": "START",
 "timestamp": "2024-12-05T10:00:00Z"
}

2. Progress Events (Optional, during execution)

{
 "eventType": "LOG",
 "timestamp": "2024-12-05T10:02:00Z",
 "contextData": {
 "message": "Extracted 50,000 of 100,000 rows",
 "progress": "50%"
 }
}

3. End Event (When job completes successfully)

{
 "eventType": "END",
 "timestamp": "2024-12-05T10:05:00Z",
 "contextData": {
 "rowsProcessed": 100000,
 "duration": "5 minutes"
 }
}

4. Error Event (If something fails)

{
 "eventType": "FAILED",
 "timestamp": "2024-12-05T10:03:00Z",
 "alert": "ERROR",
 "contextData": {
 "error": "Connection timeout to Athena",
 "details": "Network unreachable",
 "retryAfter": "30 seconds"
 }
}

Complete Event Sequence Example

For Extract Job (span 5001):

POST /torch-pipeline/api/pipelines/spans/5001/events
# Event: START at 10:00

POST /torch-pipeline/api/pipelines/spans/5001/events
# Event: LOG "50% complete" at 10:02

POST /torch-pipeline/api/pipelines/spans/5001/events
# Event: END at 10:05

For Transform Job (span 5002):

POST /torch-pipeline/api/pipelines/spans/5002/events
# Event: START at 10:05

POST /torch-pipeline/api/pipelines/spans/5002/events
# Event: END at 10:20

For Load Job (span 5003):

POST /torch-pipeline/api/pipelines/spans/5003/events
# Event: START at 10:20

POST /torch-pipeline/api/pipelines/spans/5003/events
# Event: END at 10:30

Pro Tip: The more events you log, the better visibility you'll have when debugging issues later!

Step 8: Mark Run Completion

When all jobs finish, update the run status.

API Call

PUT /torch-pipeline/api/pipelines/runs/109133

Path Parameters:

Request (Success)

{
 "run": {
 "status": "COMPLETED"
 }
}

Request (Failure)

{
 "run": {
 "status": "FAILED",
 "result": "FAILED"
 }
}

Status Changes:

• Success: STARTED → COMPLETED

• Failure: STARTED → FAILED

API Call Summary

You used 6 APIs:

1. PUT /pipelines - Created pipeline → Got pipeline.id = 15

2. POST /pipelines/15/runs - Created run → Got run.id = 109133

3. PUT /pipelines/15/jobs - Created 3 jobs

4. POST /pipelines/runs/109133/spans - Created 4 spans

5. PUT /pipelines/runs/109133 - Updated run status (2x: START, COMPLETE)

6. POST /pipelines/spans/:spanId/events - Recorded events (multiple times)

Troubleshooting