DolphinScheduler 集成：工作流编排引擎
Tags: #DolphinScheduler #Workflow #Scheduling #DAG #Orchestration #智策平台
Coomia发布于 2025年7月29日9 分钟阅读
分享本文Twitter / X
“系列：S3 数据基座 · 第 19 篇 | 难度：高级 | 阅读时间：20 分钟
DolphinScheduler 集成：工作流编排引擎
Tags: #DolphinScheduler #Workflow #Scheduling #DAG #Orchestration #智策平台
#TL;DR
Pipeline DSL 定义了数据处理的逻辑，而 DolphinScheduler 负责"何时执行、以什么频率执行、失败如何处理"。coomia-dip 平台集成 Apache DolphinScheduler 作为工作流编排引擎，实现 Pipeline 的定时调度、依赖管理、事件触发和全生命周期监控。本文完整解析 DolphinScheduler 与 Pipeline DSL 的集成架构、工作流 DAG 定义、Cron 调度策略、任务依赖与优先级管理、事件驱动调度、告警与重试机制，以及与 Ontology 操作的深度集成。
#1. 为什么选择 DolphinScheduler
#1.1 调度引擎对比
Code
调度引擎选型对比：
┌─────────────────┬──────────┬──────────┬──────────┬──────────┐
│ 特性 │ Airflow │ Dolphin │ Temporal │ Oozie │
├─────────────────┼──────────┼──────────┼──────────┼──────────┤
│ 去中心化 │ ✗ │ ✓ │ ✓ │ ✗ │
│ 多租户 │ 有限 │ ✓ │ ✓ │ ✗ │
│ 可视化 DAG 编辑 │ ✗ │ ✓ │ ✗ │ ✗ │
│ 资源隔离 │ ✗ │ ✓ │ ✓ │ 有限 │
│ 高可用 │ 需额外配置│ 原生 │ 原生 │ 需 ZK │
│ Java 生态集成 │ Python优先│ ✓ │ ✓ │ ✓ │
│ 复杂依赖管理 │ ✓ │ ✓ │ ✓ │ ✗ │
│ 社区活跃度 │ 高 │ 高 │ 高 │ 低 │
│ 学习成本 │ 中 │ 低 │ 中 │ 高 │
└─────────────────┴──────────┴──────────┴──────────┴──────────┘
选择 DolphinScheduler 理由：
1. 去中心化架构，单点故障不影响任务执行
2. 原生多租户支持，适合平台化场景
3. 可视化 DAG 编辑，降低运维门槛
4. Java 生态原生集成（Flink、Spark 任务类型内置）
#2. 集成架构
#2.1 整体架构
Code
DolphinScheduler 集成架构：
┌──────────────────────────────────────────────┐
│ coomia-dip Control Layer │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Pipeline │ │ Schedule │ │ Monitor │ │
│ │ Registry │ │ Manager │ │ Service │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ │ │
│ │ gRPC │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────┐ │
│ │ DolphinScheduler Adapter │ │
│ │ (Pipeline DSL → DS Workflow 转换) │ │
│ └──────────────────────────────────────┘ │
└──────────────────────┬───────────────────────┘
│ REST API
▼
┌──────────────────────────────────────────────┐
│ Apache DolphinScheduler │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ API Server│ │ Master │ │ Worker │ │
│ │ │ │ Server │ │ Server │ │
│ └──────────┘ └──────────┘ └──────────┘ │
│ ┌──────────┐ ┌──────────┐ │
│ │ Alert │ │ Logger │ │
│ │ Server │ │ Server │ │
│ └──────────┘ └──────────┘ │
└──────────────────────────────────────────────┘
#2.2 Pipeline → Workflow 转换
Python
class PipelineToWorkflowConverter:
"""将 Pipeline DSL 转换为 DolphinScheduler 工作流"""
def convert(self, pipeline: Pipeline) -> DSWorkflow:
workflow = DSWorkflow(
name=pipeline.name,
description=pipeline.description,
tenant=pipeline.owner,
)
# 阶段 1：数据摄入任务
ingest_task = self._create_ingest_task(pipeline)
workflow.add_task(ingest_task)
# 阶段 2：数据转换任务
transform_task = self._create_transform_task(pipeline)
workflow.add_task(transform_task)
workflow.add_dependency(ingest_task, transform_task)
# 阶段 3：数据写入任务
sink_task = self._create_sink_task(pipeline)
workflow.add_task(sink_task)
workflow.add_dependency(transform_task, sink_task)
# 阶段 4：质量检查任务
quality_task = self._create_quality_check_task(pipeline)
workflow.add_task(quality_task)
workflow.add_dependency(sink_task, quality_task)
return workflow
def _create_ingest_task(self, pipeline: Pipeline) -> DSTask:
if isinstance(pipeline.source, FlinkCDCSource):
return DSTask(
name=f"{pipeline.name}-ingest",
type="FLINK",
params={
"programType": "SQL",
"mainArgs": pipeline.source.compile_flink_sql(),
"flinkVersion": "1.18",
"deployMode": "cluster",
}
)
elif isinstance(pipeline.source, FileSource):
return DSTask(
name=f"{pipeline.name}-ingest",
type="PYTHON",
params={
"rawScript": self._generate_file_ingest_script(pipeline),
"pythonPath": "/opt/coomia-dip/python/bin/python3",
}
)
#3. 调度策略
#3.1 Cron 调度
Python
class ScheduleManager:
"""调度管理器"""
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,
start_time=schedule.start_time,
end_time=schedule.end_time,
failure_strategy=schedule.failure_strategy,
warning_type=schedule.warning_type,
)
elif schedule.type == 'interval':
cron = self._interval_to_cron(schedule.interval)
return DSSchedule(
workflow_name=pipeline_name,
cron=cron,
timezone=schedule.timezone,
)
elif schedule.type == 'event':
return self._create_event_trigger(
pipeline_name, schedule.event_config
)
# 使用示例
schedule_manager.create_schedule("equipment-sync", ScheduleConfig(
type='cron',
cron_expression='0 0 * * * ?', # 每小时
timezone='Asia/Shanghai',
failure_strategy='CONTINUE', # 失败后继续下一次
warning_type='EMAIL',
))
schedule_manager.create_schedule("daily-report", ScheduleConfig(
type='cron',
cron_expression='0 0 2 * * ?', # 每天凌晨 2 点
timezone='Asia/Shanghai',
failure_strategy='END', # 失败后停止
))
#3.2 依赖调度
Python
class DependencyManager:
"""任务依赖管理"""
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,
)
# 使用示例
dep_manager.define_dependencies("customer-analysis", [
Dependency(
upstream_workflow="customer-sync",
condition="SUCCESS", # 上游成功后才执行
),
Dependency(
upstream_workflow="transaction-sync",
condition="SUCCESS",
),
])
Code
依赖调度示例：
[customer-sync] [transaction-sync]
│ SUCCESS │ SUCCESS
└────────┐ ┌───────┘
▼ ▼
[customer-analysis]
│ SUCCESS
▼
[daily-report]
│ SUCCESS
▼
[email-notification]
#3.3 事件驱动调度
Python
class EventTriggerManager:
"""事件触发调度管理"""
async def on_ontology_change(
self, event: OntologyChangeEvent
):
"""Ontology 数据变更触发工作流"""
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,
'changed_properties': event.changed_properties,
'trigger_time': datetime.utcnow().isoformat(),
}
)
# 事件触发配置
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, # 60秒内的多次变更合并为一次触发
))
#4. 任务类型映射
#4.1 内置任务类型
Code
Pipeline 操作到 DS 任务类型映射：
┌──────────────────┬──────────────┬────────────────────┐
│ Pipeline 操作 │ DS 任务类型 │ 说明 │
├──────────────────┼──────────────┼────────────────────┤
│ mysql_cdc source │ FLINK │ Flink CDC SQL Job │
│ kafka source │ FLINK │ Flink Kafka Job │
│ file source │ PYTHON │ Python 脚本读取 │
│ transform chain │ FLINK/PYTHON │ 根据引擎选择 │
│ ontology sink │ PYTHON │ SDK 写入 Ontology │
│ doris sink │ SQL │ Doris SQL 写入 │
│ quality check │ PYTHON │ 数据质量检测脚本 │
│ notification │ HTTP │ Webhook 通知 │
└──────────────────┴──────────────┴────────────────────┘
#5. 监控与告警
#5.1 工作流状态监控
Python
class WorkflowMonitor:
"""工作流状态监控"""
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,
recent_history=[
InstanceSummary(
start_time=inst.start_time,
end_time=inst.end_time,
state=inst.state,
duration_seconds=inst.duration,
)
for inst in instances
],
success_rate=self._calculate_success_rate(instances),
avg_duration=self._calculate_avg_duration(instances),
)
async def check_sla_violations(self):
"""检测 SLA 违规"""
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 "
f"SLA duration: {last_run.duration}s > "
f"{sla.max_duration}s"
)
#5.2 告警配置
Python
# 告警配置示例
alert_config = AlertConfig(
workflow_name="equipment-sync",
rules=[
AlertRule(
condition="FAILURE",
channels=["email", "dingtalk"],
recipients=["data-team@company.com"],
message_template="Pipeline {workflow} failed at {time}: {error}",
),
AlertRule(
condition="SLA_VIOLATION",
channels=["dingtalk", "sms"],
recipients=["oncall@company.com"],
message_template="SLA violation: {workflow} took {duration}s (limit: {sla_limit}s)",
),
AlertRule(
condition="SUCCESS",
channels=["email"],
recipients=["data-team@company.com"],
frequency="daily_summary", # 每天汇总一次，不是每次成功都发
),
]
)
#6. 资源管理
#6.1 Worker 分组
Code
Worker 分组策略：
┌──────────────────────────────────────────┐
│ DolphinScheduler │
│ │
│ Worker Group: flink-workers │
│ ┌──────┐ ┌──────┐ ┌──────┐ │
│ │ W1 │ │ W2 │ │ W3 │ │
│ │ 8C32G│ │ 8C32G│ │ 8C32G│ │
│ └──────┘ └──────┘ └──────┘ │
│ 用途：Flink CDC 和流处理任务 │
│ │
│ Worker Group: python-workers │
│ ┌──────┐ ┌──────┐ │
│ │ W4 │ │ W5 │ │
│ │ 4C16G│ │ 4C16G│ │
│ └──────┘ └──────┘ │
│ 用途：Python 脚本和质量检查 │
│ │
│ Worker Group: sql-workers │
│ ┌──────┐ │
│ │ W6 │ │
│ │ 2C8G │ │
│ └──────┘ │
│ 用途：SQL 查询和数据写入 │
└──────────────────────────────────────────┘
#7. 测试策略
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 # ingest + transform + sink
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 = EventTrigger(
event_type="ENTITY_UPDATED",
entity_type="Customer",
workflow_name="risk-scoring",
)
trigger_manager.register_trigger(trigger)
event = OntologyChangeEvent(
type="ENTITY_UPDATED",
entity_type="Customer",
entity_id="c1",
changed_properties=["credit_rating"]
)
await trigger_manager.on_ontology_change(event)
# Verify workflow was triggered
async def test_dependency_execution_order(self):
# Create dependent workflows
dep_manager.define_dependencies("downstream", [
Dependency(upstream_workflow="upstream", condition="SUCCESS")
])
# Run upstream successfully
await run_workflow("upstream")
# Verify downstream was triggered
status = await monitor.get_workflow_status("downstream")
assert status.last_run.state in ('RUNNING', 'SUCCESS')
#Key Takeaways
-
DolphinScheduler 的去中心化架构适合平台化场景：Master-Worker 分离，Worker 分组管理，多租户隔离，不存在单点故障。
-
Pipeline DSL 到 DS Workflow 的自动转换降低了调度配置成本：用户只需定义 Pipeline，系统自动生成包含摄入、转换、写入和质量检查的完整工作流。
-
三种调度模式覆盖所有场景：Cron 定时调度用于批量处理，依赖调度用于多 Pipeline 编排，事件驱动用于实时响应。
-
SLA 监控和多渠道告警保障生产稳定性：自动检测执行时间违规、失败重试、降级策略和告警通知。
-
Worker 分组实现资源隔离：不同类型的任务在不同的 Worker 组上执行，避免资源竞争影响关键 Pipeline。
#Next Article
下一篇 S3-20《Transform 执行器：多引擎适配层》 将深入 Pipeline 中 Transform 步骤的多引擎执行细节，包括 Flink、Spark 和 DuckDB 的适配实现。
Tags: #DolphinScheduler #Workflow #Scheduling #DAG #Orchestration #EventDriven #CronSchedule #SLA #智策平台 #coomia-dip #数据基座