一、前言：为什么要了解 DolphinScheduler？

DolphinScheduler 的独特定位：

二、DolphinScheduler 是什么？

2.1 一句话定义

DolphinScheduler = 分布式、去中心化、易扩展的可视化工作流任务调度平台

2.2 发展历程

2.3 核心特性

三、架构详解

3.1 整体架构

3.2 核心组件职责

3.3 去中心化设计

去中心化的优势：

• 无单点故障：任一 Master 挂掉，其他 Master 自动接管
• 负载均衡：任务自动分配到空闲 Master
• 水平扩展：随时增加 Master/Worker 节点

3.4 任务执行流程

四、核心概念

4.1 概念层级图

4.2 项目与工作流

4.3 任务类型

4.4 多租户体系

租户隔离的好处：

五、实战：构建数据管道

5.1 创建第一个工作流

5.2 可视化 DAG 编辑

在 DolphinScheduler 中，你可以通过拖拽来创建工作流：

5.3 Shell 任务示例

{
  "taskType": "SHELL",
  "name": "extract_data",
  "description": "从MySQL抽取数据",
  "taskParams": {
    "resourceList": [],
    "localParams": [
      {
        "prop": "dt",
        "direct": "IN",
        "type": "VARCHAR",
        "value": "${global_dt}"
      }
    ],
    "rawScript": "#!/bin/bash\necho \"开始抽取数据，日期：${dt}\"\npython3 /scripts/extract.py --date ${dt}\necho \"抽取完成\""
  },
  "flag": "YES",
  "taskPriority": "MEDIUM",
  "workerGroup": "default",
  "failRetryTimes": 3,
  "failRetryInterval": 1,
  "timeoutFlag": "OPEN",
  "timeout": 7200
}

5.4 SQL 任务示例

{
  "taskType": "SQL",
  "name": "aggregate_daily_stats",
  "description": "计算每日统计",
  "taskParams": {
    "type": "MYSQL",
    "datasource": 1,
    "sql": "INSERT INTO daily_stats (date, total_orders, total_amount)\nSELECT \n  '${dt}' as date,\n  COUNT(*) as total_orders,\n  SUM(amount) as total_amount\nFROM orders\nWHERE DATE(create_time) = '${dt}'\nON DUPLICATE KEY UPDATE\n  total_orders = VALUES(total_orders),\n  total_amount = VALUES(total_amount);",
    "sqlType": "1",
    "preStatements": [],
    "postStatements": [],
    "displayRows": 10
  },
  "workerGroup": "default"
}

5.5 Spark 任务示例

{
  "taskType": "SPARK",
  "name": "spark_etl_job",
  "taskParams": {
    "programType": "SCALA",
    "sparkVersion": "SPARK2",
    "deployMode": "cluster",
    "mainClass": "com.company.etl.DailyETL",
    "mainJar": {
      "resourceName": "/resources/spark-jobs/etl-job.jar"
    },
    "appName": "daily_etl_${dt}",
    "driverCores": 2,
    "driverMemory": "4G",
    "numExecutors": 10,
    "executorMemory": "8G",
    "executorCores": 4,
    "mainArgs": "--date ${dt} --mode full",
    "others": "--conf spark.sql.shuffle.partitions=200"
  },
  "workerGroup": "spark"
}

5.6 依赖任务：跨工作流依赖

{
  "taskType": "DEPENDENT",
  "name": "wait_for_ods",
  "description": "等待ODS层工作流完成",
  "taskParams": {
    "dependence": {
      "relation": "AND",
      "dependTaskList": [
        {
          "relation": "AND",
          "dependItemList": [
            {
              "projectCode": 123456,
              "definitionCode": 789012,
              "depTaskCode": 0,
              "cycle": "day",
              "dateValue": "today"
            }
          ]
        }
      ]
    }
  }
}

5.7 条件分支任务

{
  "taskType": "CONDITIONS",
  "name": "check_data_size",
  "taskParams": {
    "dependence": {
      "relation": "AND",
      "dependTaskList": [
        {
          "relation": "AND", 
          "dependItemList": [
            {
              "depTaskCode": "count_task",
              "status": "SUCCESS"
            }
          ]
        }
      ]
    },
    "conditionResult": {
      "successNode": ["spark_process"],
      "failedNode": ["python_process"]
    }
  }
}

六、参数系统详解

6.1 参数类型

6.2 内置时间参数

6.3 时间函数

# 基准时间
$[yyyy-MM-dd]           # 2024-01-15

# 加减运算
$[yyyy-MM-dd+7]         # 2024-01-22（加7天）
$[yyyy-MM-dd-1]         # 2024-01-14（减1天）
$[yyyyMMdd-1]           # 20240114

# 月份操作
$[add_months(yyyy-MM-dd, -1)]  # 2023-12-15（上个月）

# 自定义格式
$[yyyyMMddHHmmss]       # 20240115120000

6.4 参数传递示例

# Task1 脚本：输出参数
#!/bin/bash
count=$(mysql -e "SELECT COUNT(*) FROM orders WHERE date='${dt}'" | tail -1)
echo "row_count=${count}"  # DolphinScheduler会解析这个输出

# Task2 脚本：使用参数
#!/bin/bash
echo "上游任务处理了 ${row_count} 条数据"
if [ ${row_count} -gt 10000 ]; then
    echo "数据量较大，启用并行处理"
fi

七、告警配置

7.1 告警渠道

7.2 告警组配置

{
  "alertGroupName": "数据开发组",
  "alertInstanceIds": "1,2,3",
  "description": "数据开发团队告警",
  "alertInstances": [
    {
      "instanceType": "DingTalk",
      "params": {
        "webhook": "https://oapi.dingtalk.com/robot/send?access_token=xxx",
        "keyword": "数据告警",
        "isAtAll": false,
        "atMobiles": "13800138000,13900139000"
      }
    },
    {
      "instanceType": "Email",
      "params": {
        "receivers": "team@company.com",
        "serverHost": "smtp.company.com",
        "serverPort": 465
      }
    }
  ]
}

7.3 告警策略

八、资源中心

8.1 资源类型

8.2 资源使用

# 在Shell任务中引用资源
#!/bin/bash

# 引用资源中心的Python脚本
python3 /资源路径/process_data.py --date ${dt}

# 引用资源中心的配置文件
source /资源路径/config.sh

九、高级特性

9.1 Worker 分组

分组的应用场景：

9.2 补数功能

补数配置：

{
  "scheduleTime": "2024-01-01 00:00:00,2024-01-15 00:00:00",
  "failureStrategy": "CONTINUE",
  "warningType": "FAILURE",
  "warningGroupId": 1,
  "execType": "COMPLEMENT_DATA",
  "runMode": "RUN_MODE_SERIAL",
  "processInstancePriority": "MEDIUM",
  "workerGroup": "default",
  "expectedParallelismNumber": 4
}

9.3 版本管理

十、与 Airflow 对比

10.1 功能对比

10.2 架构对比

10.3 选型建议

十一、生产部署

11.1 部署架构

11.2 资源规划

11.3 K8s 部署

# values.yaml 示例
master:
  replicas: 3
  resources:
    requests:
      memory: "4Gi"
      cpu: "2"
    limits:
      memory: "8Gi"
      cpu: "4"

worker:
  replicas: 5
  resources:
    requests:
      memory: "8Gi"
      cpu: "4"
    limits:
      memory: "16Gi"
      cpu: "8"

alert:
  replicas: 2

api:
  replicas: 2

zookeeper:
  replicas: 3

postgresql:
  enabled: true
  primary:
    persistence:
      size: 50Gi

十二、常见问题

12.1 FAQ

12.2 性能优化

十三、总结

13.1 DolphinScheduler 核心优势

13.2 适用场景

13.3 一句话总结

DolphinScheduler = 去中心化 + 可视化 + 多租户 + 云原生

如果你的团队不想写太多代码，需要一个开箱即用的调度平台，DolphinScheduler 是国内最佳选择之一。

如果这篇文章对你有帮助，欢迎：

• 👍 点赞：让更多人看到
• ⭐ 收藏：方便日后查阅
• 💬 评论：分享你的 DolphinScheduler 使用经验