“

Series: S3 Data Foundation · Article 19 | Level: Advanced | Reading Time: 20 min

DolphinScheduler Integration: Workflow Orchestration Engine

Tags: #DolphinScheduler #Workflow #Scheduling #DAG #Orchestration #coomia-dip

#TL;DR

Pipeline DSL defines data processing logic, while DolphinScheduler handles "when to execute, at what frequency, and what to do on failure." The coomia-dip platform integrates Apache DolphinScheduler as the workflow orchestration engine, enabling Pipeline scheduling, dependency management, event-driven triggers, and full lifecycle monitoring. This article fully dissects the DolphinScheduler integration architecture, workflow DAG definitions, Cron scheduling strategies, task dependency and priority management, event-driven scheduling, alerting and retry mechanisms, and deep integration with Ontology operations.

#1. Why DolphinScheduler

#1.1 Scheduler Comparison

Code

Scheduling Engine Comparison:

+-----------------+----------+----------+----------+----------+
| Feature          | Airflow  | Dolphin  | Temporal | Oozie    |
+-----------------+----------+----------+----------+----------+
| Decentralized    | No       | Yes      | Yes      | No       |
| Multi-tenant     | Limited  | Yes      | Yes      | No       |
| Visual DAG edit  | No       | Yes      | No       | No       |
| Resource isolate | No       | Yes      | Yes      | Limited  |
| High availability| Extra    | Native   | Native   | ZK req   |
| Java ecosystem   | Python   | Yes      | Yes      | Yes      |
| Complex deps     | Yes      | Yes      | Yes      | No       |
| Community        | High     | High     | High     | Low      |
+-----------------+----------+----------+----------+----------+

DolphinScheduler chosen because:
  1. Decentralized - no single point of failure
  2. Native multi-tenancy for platform scenarios
  3. Visual DAG editor lowers ops barrier
  4. Built-in Flink/Spark task types

#2. Integration Architecture

#2.1 Overall Architecture

Code

DolphinScheduler Integration Architecture:

+----------------------------------------------+
|             coomia-dip Control Layer           |
|  +----------+  +----------+  +----------+    |
|  | Pipeline  |  | Schedule |  | Monitor  |    |
|  | Registry  |  | Manager  |  | Service  |    |
|  +----------+  +----------+  +----------+    |
|       |              |              |          |
|       |         gRPC |              |          |
|       v              v              v          |
|  +------------------------------------------+ |
|  |     DolphinScheduler Adapter              | |
|  |  (Pipeline DSL -> DS Workflow conversion) | |
|  +------------------------------------------+ |
+-----------------------+-----------------------+
                        | REST API
                        v
+----------------------------------------------+
|          Apache DolphinScheduler              |
|  +----------+  +----------+  +----------+    |
|  | API Server|  | Master   |  | Worker   |    |
|  +----------+  +----------+  +----------+    |
|  +----------+  +----------+                   |
|  | Alert     |  | Logger   |                  |
|  +----------+  +----------+                   |
+----------------------------------------------+

#2.2 Pipeline to Workflow Conversion

Python

class PipelineToWorkflowConverter:
    """Convert Pipeline DSL to DolphinScheduler workflow"""

    def convert(self, pipeline: Pipeline) -> DSWorkflow:
        workflow = DSWorkflow(
            name=pipeline.name,
            description=pipeline.description,
            tenant=pipeline.owner,
        )

        ingest_task = self._create_ingest_task(pipeline)
        workflow.add_task(ingest_task)

        transform_task = self._create_transform_task(pipeline)
        workflow.add_task(transform_task)
        workflow.add_dependency(ingest_task, transform_task)

        sink_task = self._create_sink_task(pipeline)
        workflow.add_task(sink_task)
        workflow.add_dependency(transform_task, sink_task)

        quality_task = self._create_quality_check_task(pipeline)
        workflow.add_task(quality_task)
        workflow.add_dependency(sink_task, quality_task)

        return workflow

#3. Scheduling Strategies

#3.1 Cron Scheduling

Python

class ScheduleManager:
    """Schedule manager"""

    def create_schedule(
        self, pipeline_name: str, schedule: ScheduleConfig
    ) -> DSSchedule:
        if schedule.type == 'cron':
            return DSSchedule(
                workflow_name=pipeline_name,
                cron=schedule.cron_expression,
                timezone=schedule.timezone,
                failure_strategy=schedule.failure_strategy,
            )
        elif schedule.type == 'event':
            return self._create_event_trigger(
                pipeline_name, schedule.event_config
            )

# Examples
schedule_manager.create_schedule("equipment-sync", ScheduleConfig(
    type='cron',
    cron_expression='0 0 * * * ?',  # hourly
    timezone='Asia/Shanghai',
    failure_strategy='CONTINUE',
))

schedule_manager.create_schedule("daily-report", ScheduleConfig(
    type='cron',
    cron_expression='0 0 2 * * ?',  # daily at 2 AM
    timezone='Asia/Shanghai',
    failure_strategy='END',
))

#3.2 Dependency Scheduling

Python

class DependencyManager:
    """Task dependency management"""

    def define_dependencies(
        self, workflow_name: str, deps: list[Dependency]
    ):
        for dep in deps:
            self._ds_client.add_workflow_dependency(
                upstream=dep.upstream_workflow,
                downstream=workflow_name,
                condition=dep.condition,
            )

# Example
dep_manager.define_dependencies("customer-analysis", [
    Dependency(
        upstream_workflow="customer-sync",
        condition="SUCCESS",
    ),
    Dependency(
        upstream_workflow="transaction-sync",
        condition="SUCCESS",
    ),
])

Code

Dependency Scheduling Example:

  [customer-sync]     [transaction-sync]
       | SUCCESS            | SUCCESS
       +--------+  +-------+
                v  v
         [customer-analysis]
                | SUCCESS
                v
         [daily-report]
                | SUCCESS
                v
         [email-notification]

#3.3 Event-Driven Scheduling

Python

class EventTriggerManager:
    """Event-driven schedule management"""

    async def on_ontology_change(
        self, event: OntologyChangeEvent
    ):
        matching_triggers = self._find_matching_triggers(event)

        for trigger in matching_triggers:
            await self._ds_client.trigger_workflow(
                workflow_name=trigger.workflow_name,
                params={
                    'trigger_event': event.type,
                    'entity_type': event.entity_type,
                    'entity_id': event.entity_id,
                    'trigger_time': datetime.utcnow().isoformat(),
                }
            )

# Event trigger configuration
trigger_manager.register_trigger(EventTrigger(
    name="risk-recalculation",
    event_type="ENTITY_UPDATED",
    entity_type="Customer",
    property_filter=["credit_rating", "risk_level"],
    workflow_name="risk-scoring-pipeline",
    debounce_seconds=60,
))

#4. Task Type Mapping

Code

Pipeline Operations to DS Task Types:

+------------------+--------------+--------------------+
| Pipeline Op       | DS Task Type  | Description         |
+------------------+--------------+--------------------+
| mysql_cdc source | FLINK        | Flink CDC SQL Job  |
| kafka source     | FLINK        | Flink Kafka Job    |
| file source      | PYTHON       | Python script read |
| transform chain  | FLINK/PYTHON | Based on engine    |
| ontology sink    | PYTHON       | SDK writes to Onto |
| doris sink       | SQL          | Doris SQL write    |
| quality check    | PYTHON       | Data quality script|
| notification     | HTTP         | Webhook notify     |
+------------------+--------------+--------------------+

#5. Monitoring and Alerting

#5.1 Workflow Status Monitoring

Python

class WorkflowMonitor:
    """Workflow status monitor"""

    async def get_workflow_status(
        self, workflow_name: str
    ) -> WorkflowStatus:
        instances = await self._ds_client.list_instances(
            workflow_name=workflow_name, limit=10
        )

        return WorkflowStatus(
            workflow_name=workflow_name,
            last_run=instances[0] if instances else None,
            success_rate=self._calculate_success_rate(instances),
            avg_duration=self._calculate_avg_duration(instances),
        )

    async def check_sla_violations(self):
        for workflow in self._registered_workflows:
            sla = workflow.sla_config
            if sla is None:
                continue

            status = await self.get_workflow_status(workflow.name)
            last_run = status.last_run

            if last_run and last_run.duration > sla.max_duration:
                await self._alert_service.send_alert(
                    AlertType.SLA_VIOLATION,
                    message=(
                        f"Workflow '{workflow.name}' exceeded SLA: "
                        f"{last_run.duration}s > {sla.max_duration}s"
                    )
                )

#5.2 Alert Configuration

Python

alert_config = AlertConfig(
    workflow_name="equipment-sync",
    rules=[
        AlertRule(
            condition="FAILURE",
            channels=["email", "slack"],
            recipients=["data-team@company.com"],
        ),
        AlertRule(
            condition="SLA_VIOLATION",
            channels=["slack", "pagerduty"],
            recipients=["oncall@company.com"],
        ),
        AlertRule(
            condition="SUCCESS",
            channels=["email"],
            frequency="daily_summary",
        ),
    ]
)

#6. Resource Management

Code

Worker Group Strategy:

+------------------------------------------+
|           DolphinScheduler                |
|                                           |
|  Worker Group: flink-workers              |
|  +------+ +------+ +------+              |
|  | W1   | | W2   | | W3   |              |
|  | 8C32G| | 8C32G| | 8C32G|              |
|  +------+ +------+ +------+              |
|  Purpose: Flink CDC and stream tasks      |
|                                           |
|  Worker Group: python-workers             |
|  +------+ +------+                        |
|  | W4   | | W5   |                        |
|  | 4C16G| | 4C16G|                        |
|  +------+ +------+                        |
|  Purpose: Python scripts and QA checks    |
|                                           |
|  Worker Group: sql-workers                |
|  +------+                                 |
|  | W6   |                                 |
|  | 2C8G |                                 |
|  +------+                                 |
|  Purpose: SQL queries and data writes     |
+------------------------------------------+

#7. Testing Strategy

Python

class TestDolphinSchedulerIntegration:

    def test_pipeline_to_workflow_conversion(self):
        pipeline = (Pipeline.create("test")
            .source(...).transform(...).sink(...).build())
        workflow = converter.convert(pipeline)
        assert workflow.name == "test"
        assert len(workflow.tasks) >= 3

    def test_cron_schedule_creation(self):
        schedule = schedule_manager.create_schedule(
            "test", ScheduleConfig(
                type='cron', cron_expression='0 0 * * * ?'
            )
        )
        assert schedule.cron == '0 0 * * * ?'

    async def test_event_trigger(self):
        trigger_manager.register_trigger(EventTrigger(
            event_type="ENTITY_UPDATED",
            entity_type="Customer",
            workflow_name="risk-scoring",
        ))
        event = OntologyChangeEvent(
            type="ENTITY_UPDATED",
            entity_type="Customer",
            entity_id="c1",
        )
        await trigger_manager.on_ontology_change(event)

    async def test_dependency_execution(self):
        dep_manager.define_dependencies("downstream", [
            Dependency(upstream_workflow="upstream", condition="SUCCESS")
        ])
        await run_workflow("upstream")
        status = await monitor.get_workflow_status("downstream")
        assert status.last_run.state in ('RUNNING', 'SUCCESS')

#Key Takeaways

DolphinScheduler's decentralized architecture suits platform scenarios: Master-Worker separation, Worker grouping, multi-tenant isolation, no single point of failure.
Automatic Pipeline DSL to DS Workflow conversion reduces scheduling config cost: users define only the Pipeline, and the system auto-generates complete workflows including ingest, transform, sink, and quality check tasks.
Three scheduling modes cover all scenarios: Cron for batch processing, dependency scheduling for multi-pipeline orchestration, event-driven for real-time response.
SLA monitoring and multi-channel alerting ensure production stability: automatic duration violation detection, failure retry, degradation strategies, and alert notifications.
Worker grouping achieves resource isolation: different task types run on different Worker groups, preventing resource contention from impacting critical pipelines.

#Next Article

Next up: S3-20 "Transform Executor: Multi-Engine Adapter Layer" will dive into the multi-engine execution details of Pipeline Transform steps, including Flink, Spark, and DuckDB adapter implementations.

Tags: #DolphinScheduler #Workflow #Scheduling #DAG #Orchestration #EventDriven #CronSchedule #SLA #coomia-dip #DataFoundation