“

Series: S8 Technology Deep Dives · Article 10 | Level: Advanced | Reading Time: 20 min

DolphinScheduler Deep Dive: DAG Scheduling Engine and Data Pipeline Orchestration

#TL;DR

DolphinScheduler is the batch processing scheduling core of coomia-dip's Pipeline & Orchestration (Pipeline & Orchestration Layer), responsible for orchestrating ETL pipelines, data quality checks, scheduled aggregations, and other DAG workflows
This article deeply analyzes DolphinScheduler's Master/Worker architecture, DAG parsing and task dispatch mechanisms, fault tolerance strategies, and integration with coomia-dip's Ontology layer
Covers multi-tenancy isolation, resource management, alert integration, and DolphinScheduler 3.x's API-driven workflow definitions

#Table of Contents

DolphinScheduler's Role in coomia-dip
Master-Worker Architecture
DAG Parsing and Task Scheduling
Task Types and Plugin System
Fault Tolerance and Retry Mechanisms
Multi-Tenancy and Resource Isolation
Integration with coomia-dip Ontology
API-Driven Workflow Definition
Alerting and Monitoring Integration
Performance Tuning and Production Practices
Key Takeaways

#1. DolphinScheduler's Role in coomia-dip

#1.1 DolphinScheduler vs Temporal Division of Labor

coomia-dip uses both DolphinScheduler and Temporal, with clear responsibility boundaries:

Dimension	DolphinScheduler	Temporal
Use case	Batch ETL, scheduled jobs	Real-time event-driven, long-running workflows
Trigger	Cron / Manual / Dependency	API / Event / Schedule
Task granularity	Coarse (Shell/SQL/Flink Job)	Fine (function-level Activity)
DAG definition	Visual drag-and-drop + JSON	Code-defined
User audience	Data engineers	Developers
coomia-dip Layer	F (Pipeline)	E (Agent Runtime)

#1.2 Typical Use Cases

Code

DolphinScheduler-managed pipelines:

1. Bronze Layer Data Ingestion
   ┌──────────┐    ┌──────────┐    ┌──────────┐
   │ MySQL CDC│───→│ Kafka    │───→│ Iceberg  │
   │ Full Sync│    │ Topic    │    │ Bronze   │
   └──────────┘    └──────────┘    └──────────┘

2. Silver Layer Data Cleansing
   ┌──────────┐    ┌──────────┐    ┌──────────┐
   │ Iceberg  │───→│ Flink    │───→│ Iceberg  │
   │ Bronze   │    │ SQL Job  │    │ Silver   │
   └──────────┘    └──────────┘    └──────────┘

3. Gold Layer Aggregation
   ┌──────────┐    ┌──────────┐    ┌──────────┐
   │ Iceberg  │───→│ Trino    │───→│ Doris    │
   │ Silver   │    │ Query    │    │ Gold     │
   └──────────┘    └──────────┘    └──────────┘

#2. Master-Worker Architecture

#2.1 Core Components

Code

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

#2.2 Master Server Responsibilities

DAG parsing: Parses workflow definitions into DAG graphs
Task scheduling: Dispatches tasks based on dependencies and priorities
Fault management: Detects Worker failures, reassigns tasks
Slot management: Tracks available slots on each Worker

Java

// Master scheduling loop (simplified)
public class MasterSchedulerThread implements Runnable {
    @Override
    public void run() {
        while (isRunning) {
            // 1. Fetch commands to schedule from DB
            List<Command> commands = commandService.findCommandPage(masterSlots);

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

                // 3. Get ready tasks (all predecessors completed)
                List<TaskNode> readyTasks = getReadyTasks(dag, processInstance);

                // 4. Dispatch to Workers
                for (TaskNode task : readyTasks) {
                    WorkerInfo selectedWorker = selectWorker(task);
                    dispatchTask(task, selectedWorker);
                }
            }

            Thread.sleep(100);  // Scheduling interval
        }
    }
}

#2.3 Worker Server Responsibilities

Task execution: Receives and executes tasks from Master
Log collection: Collects task logs and reports them
Heartbeat reporting: Periodically reports liveness and load to Registry

Java

// Worker task execution flow
public class WorkerTaskExecuteRunnable implements Runnable {
    @Override
    public void run() {
        try {
            // 1. Parse task parameters
            TaskExecutionContext context = buildContext(taskInstance);

            // 2. Create task plugin instance
            AbstractTask task = TaskPluginManager.createTask(
                taskInstance.getTaskType(), context
            );

            // 3. Initialize
            task.init();

            // 4. Execute
            task.handle();

            // 5. Report result
            reportTaskResult(task.getExitStatus());
        } catch (Exception e) {
            reportTaskResult(TaskExecutionStatus.FAILURE);
        }
    }
}

#3. DAG Parsing and Task Scheduling

#3.1 DAG Data Structure

JSON

{
  "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"]
    }
  ]
}

#3.2 Topological Sort and Parallelism

Code

DAG topological sort result:

Level 0: [task-extract]              → Parallelism 1
Level 1: [task-validate]             → Parallelism 1
Level 2: [task-transform]            → Parallelism 1
Level 3: [task-load, task-alert]     → Parallelism 2 (can run in parallel)

#3.3 Scheduling Priority

Java

// Priority calculation formula
int priority = processInstancePriority * 10 + taskInstancePriority;

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

// Scheduling queue sorted by priority
PriorityQueue<TaskInstance> queue = new PriorityQueue<>(
    Comparator.comparingInt(TaskInstance::getProcessInstancePriority)
              .thenComparingInt(TaskInstance::getTaskInstancePriority)
              .thenComparing(TaskInstance::getSubmitTime)
);

#4. Task Types and Plugin System

#4.1 Task Types Used in coomia-dip

Task Type	Purpose	coomia-dip Scenario
SHELL	Script execution	Data file processing, environment prep
SQL	Database queries	Data quality checks, aggregation
FLINK	Flink Jobs	CDC pipelines, stream processing
HTTP	REST API calls	Trigger coomia-dip APIs
PYTHON	Python scripts	ML model training, data analysis
DEPENDENT	Cross-workflow dependency	Pipeline dependency chains
SUB_PROCESS	Sub-workflow	Modular pipelines
CONDITIONS	Conditional branching	Data quality branching

#4.2 Custom Task Plugin

Java

// coomia-dip custom task plugin: 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. Call coomia-dip gRPC API to get Schema
        OntologySchema schema = ontologyClient.getSchema(
            parameters.getWorldId(),
            parameters.getObjectType()
        );

        // 2. Sync to target storage
        syncToTarget(schema, parameters.getTargetDataSource());

        // 3. Record sync result
        setExitStatusCode(TaskConstants.EXIT_CODE_SUCCESS);
    }
}

#5. Fault Tolerance and Retry Mechanisms

#5.1 Failure Types and Handling Strategies

Failure Type	Detection	Handling Strategy
Worker crash	ZooKeeper heartbeat timeout	Reassign task to another Worker
Master crash	ZooKeeper election	Standby Master takes over
Task execution failure	Non-zero exit code	Retry per retry policy
Task timeout	Timeout detection thread	Kill process + mark failed
Network partition	ZooKeeper session	Re-register after recovery

#5.2 Retry Configuration

JSON

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

#5.3 Workflow-Level Fault Tolerance

JSON

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

Strategy	Behavior
`END`	Any task failure terminates entire workflow
`CONTINUE`	Failed task marked, independent downstream tasks continue

#6. Multi-Tenancy and Resource Isolation

#6.1 coomia-dip Tenant Model

Code

DolphinScheduler Tenant Hierarchy:

Security Level
├── Admin (coomia-dip administrator)
│   └── Manages all projects and resources
├── 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

#6.2 Worker Group Resource Isolation

YAML

# Worker group configuration
worker-groups:
  - name: "high-memory"
    workers: ["worker-1", "worker-2"]
    description: "16GB+ memory for large Flink jobs"

  - name: "gpu"
    workers: ["worker-3"]
    description: "GPU nodes for ML training"

  - name: "default"
    workers: ["worker-4", "worker-5", "worker-6"]
    description: "General task execution"

#6.3 Queue Management

YAML

# YARN queue mapping
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"

#7. Integration with coomia-dip Ontology

#7.1 Ontology-Based Pipeline Templates

Python

# coomia-dip pipeline template generator
class OntologyPipelineGenerator:
    """Auto-generates DolphinScheduler pipelines from Ontology Schema"""

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

#7.2 Pipeline Execution Events Written Back to Ontology

Python

# DolphinScheduler Webhook → coomia-dip audit events
@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,
            },
        )

#8. API-Driven Workflow Definition

#8.1 DolphinScheduler 3.x REST API

Python

import httpx

class DolphinSchedulerClient:
    """DolphinScheduler REST API client"""

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

#9. Alerting and Monitoring Integration

#9.1 Alert Plugin Configuration

YAML

# coomia-dip DolphinScheduler alert configuration
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"

#9.2 Prometheus Metrics

YAML

# DolphinScheduler Prometheus metrics
metrics:
  - name: dolphinscheduler_master_running_process_count
    type: gauge
    help: "Number of currently running process instances"

  - name: dolphinscheduler_master_task_dispatch_count
    type: counter
    help: "Total task dispatch count"

  - name: dolphinscheduler_worker_task_execution_count
    type: counter
    labels: [task_type, status]
    help: "Worker task execution count"

  - name: dolphinscheduler_worker_task_execution_duration
    type: histogram
    labels: [task_type]
    help: "Task execution duration distribution"

#10. Performance Tuning and Production Practices

#10.1 Master Tuning

YAML

master:
  max-cpu-load-avg: 0.7          # CPU load ceiling
  reserved-memory: 0.3            # Reserved memory ratio
  exec-threads: 100               # Scheduling threads
  dispatch-task-number: 3         # Tasks dispatched per cycle
  host-selector: lower-weight     # Worker selection strategy
  task-commit-interval: 1000      # Task commit interval (ms)

#10.2 Worker Tuning

YAML

worker:
  exec-threads: 100               # Task execution threads
  heartbeat-interval: 10          # Heartbeat interval (seconds)
  max-cpu-load-avg: 0.75         # CPU load ceiling
  reserved-memory: 0.25           # Reserved memory ratio
  tenant-auto-create: true        # Auto-create Linux users

#10.3 Database Optimization

SQL

-- Periodically clean historical data (retain 90 days)
DELETE FROM t_ds_process_instance
WHERE start_time < DATE_SUB(NOW(), INTERVAL 90 DAY)
  AND state IN (7, 8);  -- SUCCESS, FAILURE

-- Create indexes for common queries
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);

#10.4 Resource Planning

Component	Instances	CPU	Memory	Disk
Master	2 (HA)	4 cores	8 GB	50 GB
Worker	3-10	4-8 cores	8-16 GB	100 GB
API Server	2	2 cores	4 GB	20 GB
PostgreSQL	1 (HA)	4 cores	16 GB	SSD 200 GB
ZooKeeper	3	2 cores	4 GB	SSD 50 GB

#11. Key Takeaways

Topic	Key Conclusion
Positioning	Batch DAG scheduling, complementary to Temporal
Architecture	Master scheduling + Worker execution + ZooKeeper registry
DAG	Topological sort determines execution order; same-level tasks can parallelize
Fault tolerance	Automatic Worker failure reassignment, task-level retries
Multi-tenancy	Tenant + Project + Worker Group three-level isolation
Ontology integration	Auto-generate pipeline definitions from Schema
API-driven	3.x REST API supports code-based pipeline management
Monitoring	Prometheus + custom alert plugins

“

Next up: S8-11 (already published) deeply analyzes Redis's 5 roles in coomia-dip.