DolphinScheduler 管理的管道：

1. 铜层数据采集
   ┌──────────┐    ┌──────────┐    ┌──────────┐
   │ MySQL CDC│───→│ Kafka    │───→│ Iceberg  │
   │ 全量同步  │    │ Topic    │    │ Bronze   │
   └──────────┘    └──────────┘    └──────────┘

2. 银层数据清洗
   ┌──────────┐    ┌──────────┐    ┌──────────┐
   │ Iceberg  │───→│ Flink    │───→│ Iceberg  │
   │ Bronze   │    │ SQL Job  │    │ Silver   │
   └──────────┘    └──────────┘    └──────────┘

3. 金层聚合分析
   ┌──────────┐    ┌──────────┐    ┌──────────┐
   │ Iceberg  │───→│ Trino    │───→│ Doris    │
   │ Silver   │    │ Query    │    │ Gold     │
   └──────────┘    └──────────┘    └──────────┘

┌─────────────────────────────────────────────────────┐
│                 DolphinScheduler                     │
│                                                      │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │  Master  │  │  Master  │  │  API     │          │
│  │  Server  │  │  Server  │  │  Server  │          │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘          │
│       │              │              │                │
│  ┌────┴──────────────┴──────────────┴────┐          │
│  │           ZooKeeper (Registry)         │          │
│  └────┬──────────────┬──────────────┬────┘          │
│       │              │              │                │
│  ┌────┴─────┐  ┌────┴─────┐  ┌────┴─────┐          │
│  │  Worker  │  │  Worker  │  │  Worker  │          │
│  │  Server  │  │  Server  │  │  Server  │          │
│  └──────────┘  └──────────┘  └──────────┘          │
│                                                      │
│  ┌──────────────────────────────────────┐           │
│  │     PostgreSQL (Metadata Store)       │           │
│  └──────────────────────────────────────┘           │
└─────────────────────────────────────────────────────┘

// Master 的调度循环（简化）
public class MasterSchedulerThread implements Runnable {
    @Override
    public void run() {
        while (isRunning) {
            // 1. 从 DB 获取待调度的 Command
            List<Command> commands = commandService.findCommandPage(masterSlots);

            for (Command command : commands) {
                // 2. 构建 DAG
                ProcessInstance processInstance = createProcessInstance(command);
                DAG<String, TaskNode, TaskNodeRelation> dag = buildDAG(processInstance);

                // 3. 获取就绪任务（所有前置依赖已完成）
                List<TaskNode> readyTasks = getReadyTasks(dag, processInstance);

                // 4. 分发到 Worker
                for (TaskNode task : readyTasks) {
                    WorkerInfo selectedWorker = selectWorker(task);
                    dispatchTask(task, selectedWorker);
                }
            }

            Thread.sleep(100);  // 调度间隔
        }
    }
}

// Worker 的任务执行流程
public class WorkerTaskExecuteRunnable implements Runnable {
    @Override
    public void run() {
        try {
            // 1. 解析任务参数
            TaskExecutionContext context = buildContext(taskInstance);

            // 2. 创建任务插件实例
            AbstractTask task = TaskPluginManager.createTask(
                taskInstance.getTaskType(), context
            );

            // 3. 初始化
            task.init();

            // 4. 执行
            task.handle();

            // 5. 上报结果
            reportTaskResult(task.getExitStatus());
        } catch (Exception e) {
            reportTaskResult(TaskExecutionStatus.FAILURE);
        }
    }
}

{
  "tasks": [
    {
      "id": "task-extract",
      "name": "Bronze Layer Extract",
      "type": "FLINK",
      "dependence": [],
      "params": {
        "flinkVersion": "1.18",
        "jobManagerMemory": "2g",
        "taskManagerMemory": "4g",
        "mainJar": "coomia-dip-pipeline-1.0.jar",
        "mainClass": "com.onto.pipeline.BronzeExtract"
      }
    },
    {
      "id": "task-validate",
      "name": "Data Quality Check",
      "type": "SQL",
      "dependence": ["task-extract"],
      "params": {
        "datasource": "doris-analytics",
        "sql": "SELECT COUNT(*) FROM quality_check WHERE status='FAIL'"
      }
    },
    {
      "id": "task-transform",
      "name": "Silver Layer Transform",
      "type": "FLINK",
      "dependence": ["task-validate"],
      "conditionResult": {
        "successNode": ["task-load"],
        "failedNode": ["task-alert"]
      }
    },
    {
      "id": "task-load",
      "name": "Gold Layer Load",
      "type": "SQL",
      "dependence": ["task-transform"]
    },
    {
      "id": "task-alert",
      "name": "Quality Alert",
      "type": "HTTP",
      "dependence": ["task-transform"]
    }
  ]
}

DAG 拓扑排序结果：

Level 0: [task-extract]              → 并行度 1
Level 1: [task-validate]             → 并行度 1
Level 2: [task-transform]            → 并行度 1
Level 3: [task-load, task-alert]     → 并行度 2（可并行）

// 优先级计算公式
int priority = processInstancePriority * 10 + taskInstancePriority;

// 优先级级别
public enum Priority {
    HIGHEST(0),
    HIGH(1),
    MEDIUM(2),
    LOW(3),
    LOWEST(4);
}

// 调度队列按优先级排序
PriorityQueue<TaskInstance> queue = new PriorityQueue<>(
    Comparator.comparingInt(TaskInstance::getProcessInstancePriority)
              .thenComparingInt(TaskInstance::getTaskInstancePriority)
              .thenComparing(TaskInstance::getSubmitTime)
);

// coomia-dip 自定义任务插件：Ontology Sync
@AutoService(TaskChannelFactory.class)
public class OntologySyncTaskChannelFactory implements TaskChannelFactory {

    @Override
    public String getName() {
        return "ONTOLOGY_SYNC";
    }

    @Override
    public TaskChannel create() {
        return new OntologySyncTaskChannel();
    }
}

public class OntologySyncTask extends AbstractTask {

    private final OntologySyncParameters parameters;

    @Override
    public void handle() throws TaskException {
        // 1. 调用 coomia-dip gRPC API 获取 Schema
        OntologySchema schema = ontologyClient.getSchema(
            parameters.getWorldId(),
            parameters.getObjectType()
        );

        // 2. 同步到目标存储
        syncToTarget(schema, parameters.getTargetDataSource());

        // 3. 记录同步结果
        setExitStatusCode(TaskConstants.EXIT_CODE_SUCCESS);
    }
}

{
  "task": {
    "name": "Bronze Layer Extract",
    "retryTimes": 3,
    "retryInterval": "1 minute",
    "failRetryStrategy": "RETRY_FROM_CURRENT_TASK",
    "timeout": {
      "strategy": "WARN_AND_FAILED",
      "interval": 3600,
      "enable": true
    }
  }
}

{
  "processDefinition": {
    "name": "CDC Bronze Pipeline",
    "failureStrategy": "CONTINUE",
    "processInstancePriority": "HIGH",
    "warningType": "FAILURE",
    "warningGroupId": 1
  }
}

DolphinScheduler 租户体系：

Security Level
├── Admin (coomia-dip 管理员)
│   └── 管理所有项目和资源
├── Tenant: coomia-dip-world-001
│   ├── User: data-engineer-1
│   ├── Project: bronze-pipelines
│   ├── Project: silver-pipelines
│   └── Resource: /shared/jars/
├── Tenant: coomia-dip-world-002
│   ├── User: data-engineer-2
│   └── Project: etl-pipelines
└── Tenant: coomia-dip-system
    └── Project: system-maintenance

# Worker 分组配置
worker-groups:
  - name: "high-memory"
    workers: ["worker-1", "worker-2"]
    description: "16GB+ 内存，用于 Flink 大作业"

  - name: "gpu"
    workers: ["worker-3"]
    description: "GPU 节点，用于 ML 训练"

  - name: "default"
    workers: ["worker-4", "worker-5", "worker-6"]
    description: "通用任务执行"

# YARN 队列映射
queues:
  - name: "production"
    capacity: 60%
    max-capacity: 80%
    tenant: "coomia-dip-production"

  - name: "development"
    capacity: 20%
    max-capacity: 40%
    tenant: "coomia-dip-dev"

  - name: "system"
    capacity: 20%
    max-capacity: 30%
    tenant: "coomia-dip-system"

# coomia-dip 管道模板生成器
class OntologyPipelineGenerator:
    """根据 Ontology Schema 自动生成 DolphinScheduler 管道"""

    def generate_cdc_pipeline(
        self,
        object_type: str,
        source_db: str,
        target_lakehouse: str,
    ) -> dict:
        return {
            "name": f"cdc-{object_type}-pipeline",
            "tasks": [
                {
                    "name": f"extract-{object_type}",
                    "type": "FLINK",
                    "params": self._build_flink_cdc_params(
                        object_type, source_db
                    ),
                },
                {
                    "name": f"validate-{object_type}",
                    "type": "SQL",
                    "dependence": [f"extract-{object_type}"],
                    "params": self._build_quality_check_params(object_type),
                },
                {
                    "name": f"transform-{object_type}",
                    "type": "FLINK",
                    "dependence": [f"validate-{object_type}"],
                    "params": self._build_transform_params(
                        object_type, target_lakehouse
                    ),
                },
            ],
        }

    def _build_flink_cdc_params(self, object_type: str, source_db: str) -> dict:
        schema = self.ontology_client.get_schema(object_type)
        return {
            "flinkVersion": "1.18",
            "mainClass": "com.onto.pipeline.CdcExtractor",
            "programArguments": (
                f"--source-table {schema.source_table} "
                f"--source-db {source_db} "
                f"--target-table iceberg.bronze.{object_type} "
                f"--columns {','.join(p.name for p in schema.properties)}"
            ),
        }

# DolphinScheduler Webhook → coomia-dip 审计事件
@app.post("/api/v1/dolphinscheduler/webhook")
async def handle_ds_webhook(event: DSWebhookEvent):
    if event.type == "PROCESS_INSTANCE_SUCCESS":
        await ontology_client.create_audit_event(
            event_type="PIPELINE_COMPLETED",
            source="DolphinScheduler",
            metadata={
                "process_id": event.process_instance_id,
                "process_name": event.process_name,
                "duration_seconds": event.duration,
                "task_count": event.task_count,
            },
        )
    elif event.type == "PROCESS_INSTANCE_FAILURE":
        await ontology_client.create_audit_event(
            event_type="PIPELINE_FAILED",
            source="DolphinScheduler",
            metadata={
                "process_id": event.process_instance_id,
                "error": event.error_message,
                "failed_task": event.failed_task_name,
            },
        )

import httpx

class DolphinSchedulerClient:
    """DolphinScheduler REST API 客户端"""

    def __init__(self, base_url: str, token: str):
        self.base_url = base_url
        self.headers = {"token": token}

    async def create_process_definition(
        self,
        project_code: int,
        name: str,
        task_definition_json: str,
        task_relation_json: str,
    ) -> dict:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/projects/{project_code}/process-definition",
                headers=self.headers,
                data={
                    "name": name,
                    "taskDefinitionJson": task_definition_json,
                    "taskRelationJson": task_relation_json,
                    "tenantCode": "coomia-dip",
                    "executionType": "PARALLEL",
                    "timeout": 0,
                },
            )
            return response.json()

    async def run_process(
        self,
        project_code: int,
        process_definition_code: int,
        schedule_time: str | None = None,
    ) -> dict:
        async with httpx.AsyncClient() as client:
            data = {
                "processDefinitionCode": process_definition_code,
                "failureStrategy": "CONTINUE",
                "warningType": "FAILURE",
                "scheduleTime": schedule_time,
                "startNodeList": "",
                "taskDependType": "TASK_POST",
                "runMode": "RUN_MODE_SERIAL",
                "processInstancePriority": "MEDIUM",
                "workerGroup": "default",
            }
            response = await client.post(
                f"{self.base_url}/projects/{project_code}/executors/start-process-instance",
                headers=self.headers,
                data=data,
            )
            return response.json()

    async def create_schedule(
        self,
        project_code: int,
        process_definition_code: int,
        crontab: str,
    ) -> dict:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/projects/{project_code}/schedules",
                headers=self.headers,
                data={
                    "processDefinitionCode": process_definition_code,
                    "schedule": json.dumps({
                        "startTime": "2026-01-01 00:00:00",
                        "endTime": "2099-12-31 23:59:59",
                        "crontab": crontab,
                        "timezoneId": "Asia/Shanghai",
                    }),
                    "warningType": "FAILURE",
                    "failureStrategy": "END",
                    "processInstancePriority": "MEDIUM",
                    "workerGroup": "default",
                },
            )
            return response.json()

# coomia-dip DolphinScheduler 告警配置
alert:
  plugins:
    - name: "webhook"
      config:
        url: "http://coomia-dip-api:8080/api/v1/alerts/dolphinscheduler"
        headerParams: '{"Authorization": "Bearer ${coomia-dip_TOKEN}"}'
        bodyParams: '{"source": "dolphinscheduler", "event": "${msg}"}'
        contentField: "msg"
        requestType: "POST"

    - name: "dingtalk"
      config:
        webhook: "${DINGTALK_WEBHOOK_URL}"
        keyword: "DolphinScheduler"
        isAtAll: false

  groups:
    - name: "pipeline-alerts"
      description: "数据管道告警组"
      plugins: ["webhook", "dingtalk"]

# DolphinScheduler Prometheus 指标
metrics:
  - name: dolphinscheduler_master_running_process_count
    type: gauge
    help: "当前运行中的流程实例数"

  - name: dolphinscheduler_master_task_dispatch_count
    type: counter
    help: "任务分发总数"

  - name: dolphinscheduler_worker_task_execution_count
    type: counter
    labels: [task_type, status]
    help: "Worker 任务执行计数"

  - name: dolphinscheduler_worker_task_execution_duration
    type: histogram
    labels: [task_type]
    help: "任务执行耗时分布"

# Master 配置优化
master:
  max-cpu-load-avg: 0.7          # CPU 负载上限
  reserved-memory: 0.3            # 预留内存比例
  exec-threads: 100               # 调度线程数
  dispatch-task-number: 3         # 每次调度分发的任务数
  host-selector: lower-weight     # Worker 选择策略
  task-commit-interval: 1000      # 任务提交间隔（ms）

# Worker 配置优化
worker:
  exec-threads: 100               # 任务执行线程数
  heartbeat-interval: 10          # 心跳间隔（秒）
  max-cpu-load-avg: 0.75         # CPU 负载上限
  reserved-memory: 0.25           # 预留内存比例
  tenant-auto-create: true        # 自动创建 Linux 用户

-- 定期清理历史数据（保留 90 天）
DELETE FROM t_ds_process_instance
WHERE start_time < DATE_SUB(NOW(), INTERVAL 90 DAY)
  AND state IN (7, 8);  -- SUCCESS, FAILURE

-- 创建常用查询索引
CREATE INDEX idx_process_instance_state_start
ON t_ds_process_instance (state, start_time);

CREATE INDEX idx_task_instance_process_state
ON t_ds_task_instance (process_instance_id, state);