数据源 → 抽取(Extract) → 转换(Transform) → 加载(Load) → Ontology 对象
         ↓                ↓                 ↓
       Source           Transform          Sink
       Connector        Stage              Writer

┌───────────────────────────────────────┐
│            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)   │
└───────────────────────────────────────┘

pip install ontology-sdk>=1.0.0
pip install ontology-sdk[pipeline]  # Pipeline 扩展包

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

# pipelines/employee_import.yaml
name: employee_import
display_name: 员工数据导入
description: 从 HR 系统批量导入员工数据到 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 * * *"   # 每天凌晨 2 点
  timezone: Asia/Shanghai
  retry:
    max_attempts: 3
    backoff: exponential
    initial_delay: 60s

monitoring:
  alerts:
    - type: failure
      channel: dingtalk
      recipients: [data-team]
    - type: data_quality
      threshold: 0.95   # 数据质量低于 95% 告警
      channel: email
      recipients: [data-owner]

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

pipeline = (
    PipelineBuilder("employee_import")
    .display_name("员工数据导入")
    .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="Asia/Shanghai")
    .build()
)

# 注册 Pipeline
pipeline_manager.register(pipeline)
print(f"Pipeline 已注册: {pipeline.name}")

# 手动触发
run = pipeline_manager.trigger("employee_import")
print(f"运行 ID: {run.run_id}")

# 等待完成
result = run.wait(timeout=600)
print(f"状态: {result.status}")
print(f"处理记录: {result.records_processed}")
print(f"成功: {result.records_succeeded}")
print(f"失败: {result.records_failed}")
print(f"耗时: {result.duration_seconds}s")

# 查看失败记录
if result.records_failed > 0:
    failures = pipeline_manager.get_failures(run.run_id)
    for f in failures[:10]:
        print(f"  行 {f.row_number}: {f.error_message}")
        print(f"  原始数据: {f.source_data}")

# pipelines/realtime_order_sync.yaml
name: realtime_order_sync
display_name: 实时订单同步
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: 每秒处理事件数
    - name: lag
      type: gauge
      description: 消费延迟
  alerts:
    - type: lag_threshold
      threshold: 10000
      channel: dingtalk

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

class OrderTransform(CustomTransform):
    """订单事件转换器"""

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

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

        # 客户 RID 查找（带缓存）
        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:
        """错误处理策略：skip / retry / dead_letter"""
        if isinstance(error, ValidationError):
            return "dead_letter"
        return "retry"

# 构建流式 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()
)

# 启动流式处理
pipeline_manager.start_stream(stream_pipeline)

# pipelines/incremental_project_sync.yaml
name: incremental_project_sync
display_name: 项目数据增量同步
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 * * * *"   # 每 15 分钟
  timezone: Asia/Shanghai

# 查看水位线状态
watermark = pipeline_manager.get_watermark("incremental_project_sync")
print(f"当前水位线: {watermark.value}")
print(f"上次同步: {watermark.last_updated}")
print(f"同步记录数: {watermark.records_since_initial}")

# 重置水位线（重新全量同步）
pipeline_manager.reset_watermark(
    "incremental_project_sync",
    new_value="2024-01-01T00:00:00Z"
)

# 手动设置水位线
pipeline_manager.set_watermark(
    "incremental_project_sync",
    value="2025-06-01T00:00:00Z"
)

# pipelines/customer_360_fusion.yaml
name: customer_360_fusion
display_name: 客户360数据融合
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

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",       # Schema 不匹配：拒绝
    on_transform_error="retry",     # 转换错误：重试
    on_sink_error="retry",          # 写入错误：重试
    on_source_error="fail_pipeline" # 数据源错误：终止
)

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

# 查看 DLQ 中的失败记录
dlq_records = pipeline_manager.get_dead_letters(
    pipeline_name="employee_import",
    limit=50
)

for record in dlq_records:
    print(f"失败时间: {record.failed_at}")
    print(f"错误: {record.error_message}")
    print(f"原始数据: {record.original_data}")
    print(f"重试次数: {record.retry_count}")
    print("---")

# 重新处理 DLQ 记录
reprocess_result = pipeline_manager.reprocess_dead_letters(
    pipeline_name="employee_import",
    filter={"error_type": "TransientError"}
)
print(f"重新处理: {reprocess_result.total} 条, "
      f"成功: {reprocess_result.succeeded}")

# 注册新版本
pipeline_v2 = (
    PipelineBuilder("employee_import")
    .version("2.0.0")
    .changelog("新增 manager_id 字段映射，优化去重策略")
    # ... 更新后的配置
    .build()
)
pipeline_manager.register(pipeline_v2)

# 查看版本历史
versions = pipeline_manager.list_versions("employee_import")
for v in versions:
    print(f"v{v.version}: {v.changelog} (注册于 {v.registered_at})")

# 回滚到指定版本
pipeline_manager.rollback("employee_import", target_version="1.0.0")

# 获取运行状态
status = pipeline_manager.get_status("employee_import")
print(f"当前状态: {status.state}")         # running / idle / failed
print(f"上次运行: {status.last_run_at}")
print(f"上次结果: {status.last_result}")
print(f"下次调度: {status.next_scheduled}")

# 获取运行历史
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} 条 | {run.duration}s")

# 获取指标
metrics = pipeline_manager.get_metrics("employee_import")
print(f"平均吞吐: {metrics.avg_throughput} records/s")
print(f"平均延迟: {metrics.avg_latency_ms} ms")
print(f"成功率: {metrics.success_rate:.1%}")
print(f"数据质量: {metrics.data_quality_score:.1%}")

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"
)

# 第一步：定义 Object Type（如果尚未存在）
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},
    }
)

# 第二步：构建 Pipeline
pipeline = (
    PipelineBuilder("customer_sync")
    .display_name("客户数据同步")
    .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 * * *")  # 每 4 小时
    .build()
)

# 第三步：注册并测试
platform.pipelines.register(pipeline)

# 试运行（不写入）
dry_run = platform.pipelines.dry_run("customer_sync", limit=10)
print(f"试运行结果: {dry_run.records_would_process} 条")
for sample in dry_run.sample_output[:3]:
    print(f"  {sample}")

# 正式运行
run = platform.pipelines.trigger("customer_sync")
result = run.wait(timeout=300)
print(f"同步完成: {result.records_succeeded} 条成功")