“

Series: S12 Developer Tutorials · Article 7 | Level: Intermediate | Reading Time: 15 min

Pipeline Development Guide

#TL;DR

Pipelines are the data processing core of the coomia-dip platform. This guide covers building end-to-end data pipelines using the Python SDK and YAML declarative configuration: from data source ingestion, cleansing and transformation, to Ontology object write-back. It covers three modes -- batch, streaming, and incremental -- along with error handling, monitoring, and performance tuning.

#1. Pipeline Overview

#1.1 What is a Pipeline?

A Pipeline is an automated workflow in coomia-dip that transforms raw data into Ontology objects. It runs on the Data Layer (Data Layer), built on Quarkus 3.x, coordinating with the Control Layer via gRPC.

Code

Data Source → Extract → Transform → Load → Ontology Objects
               ↓          ↓          ↓
             Source     Transform    Sink
             Connector  Stage        Writer

#1.2 Pipeline Types

Type	Use Case	Scheduling	Latency
Batch	Historical import, full sync	Scheduled/Manual	Minutes to hours
Streaming	Real-time ingestion, CDC	Continuous	Seconds to milliseconds
Incremental	Periodic delta sync	Scheduled	Minutes

#1.3 Architecture

Code

┌───────────────────────────────────────┐
│            Pipeline Manager           │
│         (Control Layer - B)           │
├───────────────────────────────────────┤
│  Pipeline Registry │ Schedule Manager │
│  Version Control   │ Dependency Graph │
└────────┬──────────────────┬───────────┘
         │ gRPC             │ gRPC
         ▼                  ▼
┌─────────────────┐ ┌──────────────────┐
│  Batch Executor │ │ Stream Executor  │
│  (Data Layer-C) │ │ (Flink/CDC)      │
├─────────────────┤ ├──────────────────┤
│ Source Connectors│ │ Kafka Consumer   │
│ Transform Engine │ │ Transform Engine │
│ Sink Writers     │ │ Sink Writers     │
└─────────────────┘ └──────────────────┘
         │                  │
         ▼                  ▼
┌───────────────────────────────────────┐
│     Storage Layer (Doris + Iceberg)   │
└───────────────────────────────────────┘

#2. Environment Setup

#2.1 Install Dependencies

Bash

pip install ontology-sdk>=1.0.0
pip install ontology-sdk[pipeline]  # Pipeline extension

#2.2 Initialization

Python

from ontology_sdk import OntoPlatform
from ontology_sdk.pipeline import PipelineBuilder, BatchPipeline

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052"
)

pipeline_manager = platform.pipelines

#3. Batch Pipeline

#3.1 YAML Declarative Definition

YAML

# pipelines/employee_import.yaml
name: employee_import
display_name: Employee Data Import
description: Batch import employee data from HR system into Ontology
version: "1.0.0"
type: batch

source:
  type: jdbc
  config:
    driver: mysql
    url: jdbc:mysql://hr-db:3306/hr_system
    username: ${HR_DB_USER}
    password: ${HR_DB_PASS}
    query: |
      SELECT
        employee_id, name, email, department_code,
        salary, hire_date, job_level, manager_id
      FROM employees
      WHERE updated_at > :last_sync_time

transform:
  stages:
    - name: clean_data
      type: data_quality
      rules:
        - field: email
          validate: email_format
          on_fail: reject
        - field: name
          validate: not_empty
          on_fail: reject
        - field: salary
          validate: range
          min: 0
          max: 10000000
          on_fail: flag

    - name: enrich
      type: mapping
      mappings:
        - source: employee_id
          target: external_id
        - source: name
          target: name
        - source: email
          target: email
        - source: department_code
          target: department
          lookup:
            type: reference
            object_type: Department
            match_field: code
            return_field: rid
        - source: salary
          target: salary
          transform: "round(value, 2)"
        - source: hire_date
          target: hire_date
          transform: "parse_date(value, 'yyyy-MM-dd')"
        - source: job_level
          target: level
          transform: |
            mapping = {
              'J1': 'junior', 'J2': 'junior',
              'M1': 'mid', 'M2': 'mid',
              'S1': 'senior', 'S2': 'senior',
              'P1': 'staff', 'P2': 'principal'
            }
            mapping.get(value, 'unknown')

    - name: dedup
      type: deduplication
      key: external_id
      strategy: latest_wins

sink:
  type: ontology
  config:
    object_type: Employee
    upsert_key: external_id
    batch_size: 500
    on_conflict: update

schedule:
  cron: "0 2 * * *"   # Daily at 2 AM
  timezone: UTC
  retry:
    max_attempts: 3
    backoff: exponential
    initial_delay: 60s

monitoring:
  alerts:
    - type: failure
      channel: slack
      recipients: [data-team]
    - type: data_quality
      threshold: 0.95
      channel: email
      recipients: [data-owner]

#3.2 Python API Definition

Python

from ontology_sdk.pipeline import (
    PipelineBuilder, JdbcSource, OntologySink,
    CleanStage, MappingStage, DedupStage
)

pipeline = (
    PipelineBuilder("employee_import")
    .display_name("Employee Data Import")
    .type("batch")
    .source(
        JdbcSource(
            driver="mysql",
            url="jdbc:mysql://hr-db:3306/hr_system",
            query="SELECT * FROM employees WHERE updated_at > :last_sync_time"
        )
    )
    .transform(
        CleanStage("clean_data")
        .validate("email", "email_format", on_fail="reject")
        .validate("name", "not_empty", on_fail="reject")
        .validate("salary", "range", min=0, max=10000000, on_fail="flag")
    )
    .transform(
        MappingStage("enrich")
        .map("employee_id", "external_id")
        .map("name", "name")
        .map("email", "email")
        .map("department_code", "department",
             lookup={"object_type": "Department", "match_field": "code", "return_field": "rid"})
        .map("salary", "salary", transform="round(value, 2)")
        .map("hire_date", "hire_date", transform="parse_date(value, 'yyyy-MM-dd')")
    )
    .transform(
        DedupStage("dedup", key="external_id", strategy="latest_wins")
    )
    .sink(
        OntologySink(
            object_type="Employee",
            upsert_key="external_id",
            batch_size=500,
            on_conflict="update"
        )
    )
    .schedule(cron="0 2 * * *", timezone="UTC")
    .build()
)

# Register the pipeline
pipeline_manager.register(pipeline)
print(f"Pipeline registered: {pipeline.name}")

#3.3 Execution and Monitoring

Python

# Manual trigger
run = pipeline_manager.trigger("employee_import")
print(f"Run ID: {run.run_id}")

# Wait for completion
result = run.wait(timeout=600)
print(f"Status: {result.status}")
print(f"Records processed: {result.records_processed}")
print(f"Succeeded: {result.records_succeeded}")
print(f"Failed: {result.records_failed}")
print(f"Duration: {result.duration_seconds}s")

# View failed records
if result.records_failed > 0:
    failures = pipeline_manager.get_failures(run.run_id)
    for f in failures[:10]:
        print(f"  Row {f.row_number}: {f.error_message}")
        print(f"  Source data: {f.source_data}")

#4. Streaming Pipeline

#4.1 Kafka Source

YAML

# pipelines/realtime_order_sync.yaml
name: realtime_order_sync
display_name: Real-time Order Sync
type: streaming

source:
  type: kafka
  config:
    bootstrap_servers: kafka:9092
    topic: order-events
    group_id: coomia-dip-order-sync
    auto_offset_reset: latest
    value_deserializer: json

transform:
  stages:
    - name: parse_event
      type: custom
      handler: |
        def transform(event):
            data = event['payload']
            return {
                'order_id': data['id'],
                'customer_rid': lookup_customer(data['customer_id']),
                'total_amount': float(data['total']),
                'status': data['status'].lower(),
                'items': data['line_items'],
                'created_at': parse_timestamp(data['created_at']),
                'updated_at': parse_timestamp(data['updated_at'])
            }

    - name: validate
      type: data_quality
      rules:
        - field: order_id
          validate: not_empty
        - field: total_amount
          validate: positive
        - field: status
          validate: enum
          values: [pending, confirmed, shipped, delivered, cancelled]

sink:
  type: ontology
  config:
    object_type: Order
    upsert_key: order_id
    batch_size: 100
    flush_interval: 5s
    on_conflict: update

monitoring:
  metrics:
    - name: throughput
      type: counter
      description: Events processed per second
    - name: lag
      type: gauge
      description: Consumer lag
  alerts:
    - type: lag_threshold
      threshold: 10000
      channel: slack

#4.2 Python Custom Transform

Python

from ontology_sdk.pipeline import StreamPipeline, KafkaSource, OntologySink
from ontology_sdk.pipeline.transforms import CustomTransform

class OrderTransform(CustomTransform):
    """Order event transformer"""

    def __init__(self, platform):
        self.platform = platform
        self._customer_cache = {}

    def transform(self, record: dict) -> dict:
        payload = record.get("payload", record)

        # Customer RID lookup with cache
        customer_id = payload["customer_id"]
        if customer_id not in self._customer_cache:
            customer = self.platform.objects.get(
                "Customer",
                filters={"external_id": customer_id}
            )
            self._customer_cache[customer_id] = customer.rid if customer else None

        return {
            "order_id": payload["id"],
            "customer_rid": self._customer_cache.get(customer_id),
            "total_amount": float(payload["total"]),
            "status": payload["status"].lower(),
            "item_count": len(payload.get("line_items", [])),
            "created_at": payload["created_at"],
        }

    def on_error(self, record: dict, error: Exception) -> str:
        """Error handling strategy: skip / retry / dead_letter"""
        if isinstance(error, ValidationError):
            return "dead_letter"
        return "retry"

# Build streaming pipeline
stream_pipeline = (
    StreamPipeline("realtime_order_sync")
    .source(KafkaSource(
        bootstrap_servers="kafka:9092",
        topic="order-events",
        group_id="coomia-dip-order-sync"
    ))
    .transform(OrderTransform(platform))
    .sink(OntologySink(
        object_type="Order",
        upsert_key="order_id",
        batch_size=100,
        flush_interval_seconds=5
    ))
    .build()
)

# Start streaming
pipeline_manager.start_stream(stream_pipeline)

#5. Incremental Pipeline

#5.1 Watermark-Based Incremental Sync

YAML

# pipelines/incremental_project_sync.yaml
name: incremental_project_sync
display_name: Project Data Incremental Sync
type: incremental

source:
  type: jdbc
  config:
    driver: postgresql
    url: jdbc:postgresql://pm-db:5432/project_mgmt
    query: |
      SELECT * FROM projects
      WHERE updated_at > :watermark
      ORDER BY updated_at ASC
    watermark:
      field: updated_at
      type: timestamp
      initial: "2024-01-01T00:00:00Z"

transform:
  stages:
    - name: map_fields
      type: mapping
      mappings:
        - source: project_id
          target: external_id
        - source: project_name
          target: name
        - source: project_status
          target: status
          transform: |
            status_map = {
              'ACTIVE': 'in_progress',
              'PLANNED': 'planning',
              'COMPLETED': 'done',
              'ON_HOLD': 'paused',
              'CANCELLED': 'cancelled'
            }
            status_map.get(value, 'unknown')
        - source: budget_amount
          target: budget
        - source: start_date
          target: start_date
        - source: end_date
          target: end_date

    - name: soft_delete_check
      type: custom
      handler: |
        def transform(record):
            if record.get('is_deleted'):
                record['_action'] = 'delete'
            return record

sink:
  type: ontology
  config:
    object_type: Project
    upsert_key: external_id
    handle_deletes: true

schedule:
  cron: "*/15 * * * *"   # Every 15 minutes
  timezone: UTC

#5.2 Watermark Management

Python

# View watermark status
watermark = pipeline_manager.get_watermark("incremental_project_sync")
print(f"Current watermark: {watermark.value}")
print(f"Last synced: {watermark.last_updated}")
print(f"Records since initial: {watermark.records_since_initial}")

# Reset watermark (re-sync from scratch)
pipeline_manager.reset_watermark(
    "incremental_project_sync",
    new_value="2024-01-01T00:00:00Z"
)

# Manually set watermark
pipeline_manager.set_watermark(
    "incremental_project_sync",
    value="2025-06-01T00:00:00Z"
)

#6. Multi-Source Data Fusion

#6.1 Multi-Source Pipeline

YAML

# pipelines/customer_360_fusion.yaml
name: customer_360_fusion
display_name: Customer 360 Data Fusion
type: batch

sources:
  - name: crm_data
    type: jdbc
    config:
      driver: mysql
      url: jdbc:mysql://crm-db:3306/crm
      query: "SELECT * FROM customers WHERE updated_at > :last_sync_time"

  - name: billing_data
    type: jdbc
    config:
      driver: postgresql
      url: jdbc:postgresql://billing-db:5432/billing
      query: |
        SELECT customer_id,
               SUM(amount) AS total_revenue,
               COUNT(*) AS transaction_count,
               MAX(transaction_date) AS last_transaction
        FROM transactions
        WHERE transaction_date > :last_sync_time
        GROUP BY customer_id

  - name: support_data
    type: api
    config:
      url: https://support-api.internal/customers/export
      method: GET
      headers:
        Authorization: "Bearer ${SUPPORT_API_TOKEN}"
      pagination:
        type: cursor
        cursor_field: next_cursor

transform:
  stages:
    - name: join_sources
      type: join
      config:
        primary: crm_data
        joins:
          - source: billing_data
            on: crm_data.customer_id = billing_data.customer_id
            type: left
          - source: support_data
            on: crm_data.customer_id = support_data.customer_id
            type: left

    - name: compute_tier
      type: custom
      handler: |
        def transform(record):
            revenue = record.get('total_revenue', 0)
            if revenue > 10000000:
                record['tier'] = 'platinum'
            elif revenue > 5000000:
                record['tier'] = 'gold'
            elif revenue > 1000000:
                record['tier'] = 'silver'
            else:
                record['tier'] = 'bronze'
            return record

    - name: compute_health_score
      type: custom
      handler: |
        def transform(record):
            scores = []
            if record.get('last_transaction'):
                days_since = (now() - record['last_transaction']).days
                scores.append(max(0, 100 - days_since))
            if record.get('open_tickets', 0) > 5:
                scores.append(30)
            elif record.get('open_tickets', 0) > 0:
                scores.append(70)
            else:
                scores.append(100)
            record['health_score'] = sum(scores) / len(scores) if scores else 50
            return record

sink:
  type: ontology
  config:
    object_type: Customer
    upsert_key: external_id
    batch_size: 200

#7. Error Handling and Retries

#7.1 Error Handling Strategies

Python

from ontology_sdk.pipeline import ErrorHandler, DeadLetterQueue

error_handler = ErrorHandler(
    max_retries=3,
    retry_backoff="exponential",
    initial_delay_seconds=5,
    dead_letter_queue=DeadLetterQueue(
        type="kafka",
        topic="pipeline-dead-letters"
    ),
    on_schema_error="reject",
    on_transform_error="retry",
    on_sink_error="retry",
    on_source_error="fail_pipeline"
)

pipeline = (
    PipelineBuilder("robust_import")
    .error_handler(error_handler)
    # ... source, transform, sink config
    .build()
)

#7.2 Dead Letter Queue Processing

Python

# View DLQ records
dlq_records = pipeline_manager.get_dead_letters(
    pipeline_name="employee_import",
    limit=50
)

for record in dlq_records:
    print(f"Failed at: {record.failed_at}")
    print(f"Error: {record.error_message}")
    print(f"Original data: {record.original_data}")
    print(f"Retry count: {record.retry_count}")
    print("---")

# Reprocess DLQ records
reprocess_result = pipeline_manager.reprocess_dead_letters(
    pipeline_name="employee_import",
    filter={"error_type": "TransientError"}
)
print(f"Reprocessed: {reprocess_result.total}, "
      f"Succeeded: {reprocess_result.succeeded}")

#8. Pipeline Version Management

#8.1 Version Control

Python

# Register new version
pipeline_v2 = (
    PipelineBuilder("employee_import")
    .version("2.0.0")
    .changelog("Added manager_id field mapping, improved dedup strategy")
    # ... updated config
    .build()
)
pipeline_manager.register(pipeline_v2)

# View version history
versions = pipeline_manager.list_versions("employee_import")
for v in versions:
    print(f"v{v.version}: {v.changelog} (registered {v.registered_at})")

# Rollback to a specific version
pipeline_manager.rollback("employee_import", target_version="1.0.0")

#9. Monitoring and Alerting

#9.1 Pipeline Dashboard

Python

# Get pipeline status
status = pipeline_manager.get_status("employee_import")
print(f"Current state: {status.state}")
print(f"Last run: {status.last_run_at}")
print(f"Last result: {status.last_result}")
print(f"Next scheduled: {status.next_scheduled}")

# Get run history
history = pipeline_manager.get_history(
    "employee_import",
    limit=10,
    since="2025-01-01"
)

for run in history:
    print(f"  {run.started_at} | {run.status} | "
          f"{run.records_processed} records | {run.duration}s")

# Get metrics
metrics = pipeline_manager.get_metrics("employee_import")
print(f"Avg throughput: {metrics.avg_throughput} records/s")
print(f"Avg latency: {metrics.avg_latency_ms} ms")
print(f"Success rate: {metrics.success_rate:.1%}")
print(f"Data quality: {metrics.data_quality_score:.1%}")

#10. Complete Example: Building a Customer Data Pipeline from Scratch

Python

from ontology_sdk import OntoPlatform
from ontology_sdk.pipeline import PipelineBuilder, JdbcSource, OntologySink

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052"
)

# Step 1: Define Object Type (if not already exists)
platform.schema.create_object_type(
    name="Customer",
    properties={
        "external_id": {"type": "string", "required": True, "indexed": True},
        "name": {"type": "string", "required": True},
        "industry": {"type": "string"},
        "region": {"type": "string"},
        "annual_revenue": {"type": "decimal"},
        "tier": {"type": "string", "enum": ["platinum", "gold", "silver", "bronze"]},
        "health_score": {"type": "integer", "min": 0, "max": 100},
    }
)

# Step 2: Build Pipeline
pipeline = (
    PipelineBuilder("customer_sync")
    .display_name("Customer Data Sync")
    .type("incremental")
    .source(JdbcSource(
        driver="mysql",
        url="jdbc:mysql://crm-db:3306/crm",
        query="SELECT * FROM customers WHERE updated_at > :watermark",
        watermark_field="updated_at",
        watermark_type="timestamp"
    ))
    .transform_mapping({
        "id": "external_id",
        "company_name": "name",
        "industry_code": ("industry", lambda v: INDUSTRY_MAP.get(v, "Other")),
        "country": ("region", lambda v: REGION_MAP.get(v, v)),
        "revenue": ("annual_revenue", lambda v: round(float(v), 2)),
    })
    .transform_custom(compute_customer_tier)
    .sink(OntologySink(
        object_type="Customer",
        upsert_key="external_id",
        batch_size=500
    ))
    .schedule(cron="0 */4 * * *")  # Every 4 hours
    .build()
)

# Step 3: Register and test
platform.pipelines.register(pipeline)

# Dry run (no writes)
dry_run = platform.pipelines.dry_run("customer_sync", limit=10)
print(f"Dry run result: {dry_run.records_would_process} records")
for sample in dry_run.sample_output[:3]:
    print(f"  {sample}")

# Production run
run = platform.pipelines.trigger("customer_sync")
result = run.wait(timeout=300)
print(f"Sync complete: {result.records_succeeded} records succeeded")

#Key Takeaways

Three pipeline modes: Batch for full sync, streaming for event-driven, incremental for periodic deltas
Declarative first: Prefer YAML declarative definitions; use Python custom transforms for complex logic
No data loss: Configure Dead Letter Queues to ensure failed records are traceable and reprocessable
Watermark mechanism: Incremental sync is based on watermarks, supporting reset and manual adjustment
Monitor first: Every pipeline should have alerting and metrics monitoring configured
Version management: Pipeline changes must be versioned and support rollback

#Next Article

Next: S12-08 Custom Function Development Guide — Learn how to write Python custom functions to extend Ontology's computation capabilities.

Tags: Pipeline ETL Data Pipeline Batch Processing Streaming Incremental Sync coomia-dip