调度与资源层

调度与资源是两条互补的问题域：DolphinScheduler 解决"时间维度"——按 cron 定时或依赖事件触发工作流，管理任务依赖、补数、重试与告警；YARN 解决"空间维度"——把集群 CPU 与内存抽象为队列与 Container，在多租户之间做配额、隔离与弹性共享。二者协同的典型链路是：DolphinScheduler 在约定时刻把 Spark/Flink 作业提交到 YARN 的指定队列，YARN 按队列容量与优先级分配 Container，作业失败由 DolphinScheduler 感知并按策略重试、告警。

Apache DolphinScheduler

DolphinScheduler（下称 DS）是 Apache 顶级开源的分布式可视化 DAG 工作流调度系统，以"去中心化 Master 集群 + Worker 分组"支撑高可用与水平扩展，是国产湖仓/数仓场景中使用最广泛的调度器之一。相对于 Crontab、Azkaban、Airflow，DS 的优势集中在：DAG 可视化编排 补数 多租户 Worker 分组 四个生产刚需上。

1. 服务架构

2. DAG 编排核心概念

3. 湖仓任务定义要点

DS 提供多种任务类型，湖仓链路中最常用的是 SHELL、SQL、SPARK、FLINK 四种。无论哪种类型，都应遵守两条规范：时间一律用内置参数（如 $[yyyy-MM-dd]）而非硬编码，保证补数可回放；任务必须幂等（Iceberg/Hudi 表按主键或分区覆盖写），保证重试安全。

SHELL 任务——封装 spark-submit（常用于带复杂参数的批作业）：

#!/bin/bash
# DS SHELL 任务：Spark 清洗作业（ODS -> DWD），日期由 DS 参数替换
set -euo pipefail

dt="$[yyyy-MM-dd]"        # 内置时间参数：补数时按区间逐日替换

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --queue lakehouse_offline \
  --name "dwd_orders_clean_${dt}" \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.sql.shuffle.partitions=800 \
  --num-executors 20 \
  --executor-cores 4 \
  --executor-memory 8g \
  --class com.lakehouse.job.DwdOrdersClean \
  hdfs:///lakehouse/jobs/dwd-orders-clean-1.0.0.jar \
  --date "${dt}"

SQL 任务——SparkSQL 数据源（适合声明式的分区覆盖写）：

-- DS SQL 任务（SparkSQL 类型）：DWD -> DWS 日聚合
SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.coalescePartitions.enabled = true;

INSERT OVERWRITE lakehouse.dws_orders_day
PARTITION (dt = '$[yyyy-MM-dd]')
SELECT user_id,
       count(*)          AS order_cnt,
       sum(order_amount) AS order_amt
FROM   lakehouse.dwd_orders
WHERE  dt = '$[yyyy-MM-dd]'
GROUP BY user_id;

FLINK 任务——流作业常驻提交：

# DS FLINK 任务类型核心参数（等价于如下 flink run 命令）
flink run \
  -m yarn-cluster \
  -yqu lakehouse_realtime \
  -yjm 2048 -ytm 4096 -ys 2 \
  -yD metrics.reporter.prom.factory.class=org.apache.flink.metrics.prometheus.PrometheusReporterFactory \
  -c com.lakehouse.stream.BinlogToIceberg \
  hdfs:///lakehouse/jobs/binlog-to-iceberg-1.0.0.jar \
  --kafka-topic ods_binlog --iceberg-table lakehouse.ods_orders

SQL 任务——JDBC 数据源（质量校验最常用）：

-- DS SQL 任务（MySQL / JDBC 类型）：把质量校验规则做成"违规明细查询"
-- 规则 1：当日分区行数不得低于昨日分区的 50%（同比波动）
-- 规则 2：主键重复率不得超过 0.1%
SELECT 'partition_cnt_drop' AS broken_rule, '$[yyyy-MM-dd]' AS dt
FROM   qc.dws_orders_cnt
WHERE  dt = '$[yyyy-MM-dd]' AND cnt < threshold * 0.5
UNION ALL
SELECT 'pk_duplicate', '$[yyyy-MM-dd]'
FROM   qc.dws_orders_dup
WHERE  dt = '$[yyyy-MM-dd]' AND dup_rate > 0.001;
-- 结果写入 qc 违规明细表后，由下游 SHELL 任务统计行数：
-- 违规行数 > 0 则 exit 1，把工作流标记失败并阻断指标产出

DEPENDENT 任务——上游工作流实例依赖（跨链路就绪保障）：

# 就绪探测 SHELL：上游分区到位才放行下游清洗（DS 内置于依赖节点后）
#!/bin/bash
set -euo pipefail
dt="$[yyyy-MM-dd]"

for i in $(seq 1 60); do
  # 检查上游 HDFS 分区的 _SUCCESS 标记（binlog 入湖作业完成时写入）
  if hdfs dfs -test -e "/warehouse/ods_orders/dt=${dt}/_SUCCESS"; then
    files=$(hdfs dfs -count "/warehouse/ods_orders/dt=${dt}" | awk '{print $2}')
    # 分区文件数大于 0 视为就绪，退出码 0 放行下游任务
    [ "${files}" -gt 0 ] && exit 0
  fi
  sleep 60
done
echo "上游分区 dt=${dt} 超过 60 分钟未就绪" >&2
exit 1    # 非零退出码触发 DS 失败重试 / 告警

4. 定时 cron 与依赖的组合策略

DS 使用 Quartz 6 域 cron 表达式（秒 分 时 日 月 周），示例：

三种触发组合的取舍：纯时间驱动最简单，但上游延迟时下游空跑或脏跑；时间 + 工作流内依赖保证链内顺序，是最常用的日批模式；时间 + 跨工作流 DEPENDENT + 就绪探测最稳，上游未就绪时任务阻塞等待而非错误运行。

5. 失败重试与告警策略

6. 湖仓场景典型 DAG

以电商湖仓的日批次为例，完整链路为：数据就绪探测 → 清洗 → 聚合 → Compaction → 质量校验 → 指标产出。DAG 中的每一环都安排了错峰时刻，避免峰值资源叠加：

Apache YARN

YARN（Yet Another Resource Negotiator）是 Hadoop 生态的资源管理中枢。湖仓集群中 Spark、Flink、Hive/Tez 等计算引擎都以 Client/AM 模式向 YARN 申请资源，因此队列设计与容量规划直接决定了多引擎、多租户能否和平共处。

1. 核心架构

2. 队列模型与资源隔离：capacity-scheduler.xml

湖仓集群的典型队列划分是"实时流队列 + 离线批队列 + 兜底队列"。以下配置将集群切分为三个一级队列：realtime（Flink 常驻流作业，高保障）、offline（Spark/Hive 离线批，弹性最大）、default（临时查询与开发调试兜底）：

<configuration>

  <!-- 一级队列：实时流 / 离线批 / 兜底，capacity 之和必须为 100 -->
  <property>
    <name>yarn.scheduler.capacity.root.queues</name>
    <value>realtime,offline,default</value>
  </property>

  <!-- ========== realtime：Flink 常驻流作业，高保障队列 ========== -->
  <property>
    <name>yarn.scheduler.capacity.root.realtime.capacity</name>
    <value>30</value>
    <!-- 正常时段最低保障 30% 集群资源 -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.realtime.maximum-capacity</name>
    <value>50</value>
    <!-- 空闲时最多借用至 50%，防止单队列吃满集群 -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.realtime.acl_submit_applications</name>
    <value>flink,rt_user</value>
    <!-- 仅允许流平台账号提交，防止批任务挤占 -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.realtime.acl_administer_queue</name>
    <value>yarn,flink</value>
  </property>
  <property>
    <!-- 实时队列中的作业不允许被抢占（关键作业保障） -->
    <name>yarn.scheduler.capacity.root.realtime.disable_preemption</name>
    <value>true</value>
  </property>

  <!-- ========== offline：Spark / Hive 离线批，弹性最大 ========== -->
  <property>
    <name>yarn.scheduler.capacity.root.offline.capacity</name>
    <value>60</value>
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.offline.maximum-capacity</name>
    <value>80</value>
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.offline.user-limit-factor</name>
    <value>2</value>
    <!-- 单用户最多占用队列容量的 2 倍（借用弹性时） -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.offline.acl_submit_applications</name>
    <value>spark,batch_user</value>
  </property>

  <!-- ========== default：临时查询与开发调试兜底 ========== -->
  <property>
    <name>yarn.scheduler.capacity.root.default.capacity</name>
    <value>10</value>
  </property>
  <property>
    <name>yarn.scheduler.capacity.root.default.maximum-capacity</name>
    <value>30</value>
    <!-- 限制临时队列的毛刺，避免挤占关键链路 -->
  </property>

  <!-- ========== 全局调度行为 ========== -->
  <property>
    <name>yarn.scheduler.capacity.maximum-applications</name>
    <value>10000</value>
    <!-- 集群并发应用上限，防止小任务风暴拖垮 RM -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.maximum-am-resource-percent</name>
    <value>0.1</value>
    <!-- AM 最多占用 10% 资源，给 Task Container 留余量 -->
  </property>
  <property>
    <name>yarn.scheduler.capacity.node-locality-delay</name>
    <value>40</value>
    <!-- 放弃节点本地性调度前等待的心跳轮数 -->
  </property>
</configuration>

3. 标签调度：节点级物理隔离

队列隔离是"逻辑配额"，仍共享同一批节点；当实时流作业对延迟极度敏感（不希望与重批任务争抢磁盘/网络）时，用 Node Label 把集群划分为物理分组，让 Flink 作业独占高配机器：

# 1) yarn-site.xml：开启节点标签（RM 侧）
<property>
  <name>yarn.node-labels.enabled</name>
  <value>true</value>
</property>
<property>
  <name>yarn.node-labels.fs-store.root-dir</name>
  <value>hdfs:///yarn/node-labels</value>
</property>

# 2) 创建标签并绑定节点（无需重启集群）
yarn rmadmin -addToClusterNodeLabels "realtime,offline"
yarn rmadmin -replaceLabelsOnNode \
  "node-01.foo.com=realtime,node-02.foo.com=realtime,node-11.foo.com=offline,node-12.foo.com=offline"

# 3) capacity-scheduler.xml：把标签授权给队列
<property>
  <name>yarn.scheduler.capacity.root.realtime.accessible-node-labels</name>
  <value>realtime</value>
</property>
<property>
  <name>yarn.scheduler.capacity.root.realtime.accessible-node-labels.realtime.capacity</name>
  <value>100</value>
  <!-- realtime 队列独占 realtime 标签节点 100% 容量 -->
</property>
<property>
  <name>yarn.scheduler.capacity.root.offline.accessible-node-labels</name>
  <value>offline</value>
</property>
<property>
  <name>yarn.scheduler.capacity.root.offline.accessible-node-labels.offline.capacity</name>
  <value>100</value>
</property>

4. 核心容量参数规划公式

# ---------- 节点与集群容量 ----------
单节点可分配内存 = 物理内存 × 0.75 − OS / 监控 Agent 预留（通常 4 ~ 8GB）
单节点可分配 vcores = 物理核数（超线程开启且 CPU 不敏感时可放宽至 1.5 倍）
集群总容量 = Σ 各节点可分配资源

# ---------- 队列配额 ----------
队列 capacity（保底）   = 队列高峰稳态需求 ÷ 集群总容量 × 100
队列 maximum-capacity = 历史峰值毛刺需求 ÷ 集群总容量 × 100 + 20% 弹性缓冲

# ---------- Container 规格 ----------
单 Container 内存 = 执行器 JVM 堆 × (1 + 10% ~ 15% 堆外 overhead)
memory : vcore 经验比 = 4GB ~ 8GB : 1（重 shuffle 批作业取高值）

# ---------- 校验 ----------
稳态并发 Container 总内存 ≤ offline 队列 capacity × 集群总内存（保底可满足）
高峰并发 Container 总内存 ≤ offline 队列 maximum-capacity × 集群总内存（弹性可消化）

资源隔离落地

队列隔离、标签隔离、K8s 混合部署是湖仓生产环境三种主流的资源隔离方案，隔离强度与成本依次递增：

关键作业保障：抢占配置

当高保障队列（如 realtime）被低保障队列占用了保底容量时，YARN 抢占机制会先向被抢占队列发出通知（SIGTERM），超过宽限期后强制回收（SIGKILL）。核心配置：

<!-- 全局开启抢占（capacity-scheduler.xml） -->
<property>
  <name>yarn.scheduler.capacity.preemption.enabled</name>
  <value>true</value>
</property>
<property>
  <name>yarn.scheduler.capacity.preemption.monitor_interval</name>
  <value>15s</value>
  <!-- 抢占监测周期 -->
</property>
<property>
  <name>yarn.scheduler.capacity.preemption.max_wait_before_kill</name>
  <value>15000</value>
  <!-- 被抢占 Container 收到通知后的自行退出宽限期（ms） -->
</property>
<property>
  <name>yarn.scheduler.capacity.preemption.total_preemption_per_round</name>
  <value>0.2</value>
  <!-- 每轮最多回收全集群 20% 资源，避免"抢占风暴" -->
</property>

<!-- 队列级开关：<queue-path> 替换为实际队列路径 -->
<property>
  <name>yarn.scheduler.capacity.<queue-path>.disable_preemption</name>
  <value>false</value>
  <!-- false = 本队列允许被高保障队列抢占；关键队列（如 realtime）设 true 免被抢 -->
</property>

调度与湖仓的衔接

湖仓一体引入开放表格式后，表的维护作业（Compaction、快照过期、Manifest 合并）必须像业务作业一样被调度化、被监控——它们不跑，表的读写性能就会慢性劣化。这正是调度层与存储层的衔接点。

1. Compaction 作业调度化

以 Iceberg 为例，三个维护性存储过程建议固定在日批次末尾执行（凌晨低峰）：

-- 1) 合并小数据文件：binpack 策略，目标 512MB / 文件
CALL lakehouse.system.rewrite_data_files(
  table    => 'lakehouse.dws_orders_day',
  strategy => 'binpack',
  options  => map('target-file-size-bytes', '536870912')
);

-- 2) 合并 manifest 元数据文件，降低查询计划阶段的扫描量
CALL lakehouse.system.rewrite_manifests(
  table => 'lakehouse.dws_orders_day'
);

-- 3) 过期快照：保留最近 7 天与最后 100 个快照
--    $[yyyy-MM-dd-7] 为 DS 内置时间参数，自动替换为 7 天前日期
CALL lakehouse.system.expire_snapshots(
  table       => 'lakehouse.dws_orders_day',
  older_than  => TIMESTAMP '$[yyyy-MM-dd-7] 00:00:00',
  retain_last => 100
);

在 DolphinScheduler 中落地为 SHELL 任务（便于参数化与失败告警），接入日批次 DAG：

#!/bin/bash
# DS SHELL 任务：Iceberg 表维护（快照过期 + 小文件合并）
set -euo pipefail

TABLE="lakehouse.dws_orders_day"
EXPIRE_DT="$[yyyy-MM-dd-7]"       # 保留 7 天快照

spark-sql \
  --master yarn \
  --queue lakehouse_offline \
  --name "iceberg_maintain_${TABLE}" \
  --num-executors 10 \
  --executor-cores 4 \
  --executor-memory 8g \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.lakehouse=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.lakehouse.type=hive \
  -e "CALL lakehouse.system.expire_snapshots(table => '${TABLE}', older_than => TIMESTAMP '${EXPIRE_DT} 00:00:00', retain_last => 100); \
      CALL lakehouse.system.rewrite_data_files(table => '${TABLE}', strategy => 'binpack');"

2. 数据就绪：事件驱动 vs 纯时间驱动

调优要点

1. 作业错峰

凌晨批次的失败大头不是代码，而是"峰值叠加"——多个大任务同一分钟提交，队列瞬间打满。错峰排布示例：

经验法则：同一时刻并行任务的峰值申请之和，不应超过 offline 队列 maximum-capacity 对应容量的 80%，剩余 20% 留给重试与临时查询。

2. 大任务拆分

• 触发条件：单任务时长超过 1 小时、或申请超过 200 executor、或处理超过一个月的全量分区——三条中满足任意一条即拆。
• 按分区拆分最常见：DS 中为每个分区生成并发任务实例（传参 ${p_date}），补数并行度即分区数；主键哈希分片适合无天然分区的表。
• 拆分后收益：失败重试粒度从"整任务"降到"单分片"，单点数据倾斜的影响面也被隔离。

3. Worker 容量

# DS 3.x application.yaml 关键字段（不同版本命名略有差异，以部署文档为准）
worker:
  exec-threads: 100          # Worker 并发处理命令的线程数
  exec-task-num: 20          # 单 Worker 同时执行任务数上限
  heartbeat-interval: 10s    # 心跳周期，Master 依此判定存活

# Worker 容量经验公式
安全并发任务数 = floor((Worker 物理内存 − 4GB 预留) ÷ 单任务客户端峰值内存)
示例：64GB Worker，spark-submit 客户端峰值约 2GB → 并发 ≤ 30，exec-task-num 取 24 留 20% 余量
过载信号：DS 任务"提交成功但长时间无日志输出"、Worker 端任务队列等待时间 > 2 × 任务历史 P50 时长

4. YARN 内存与 vcore 比例

两个高频坑：其一，忘记 spark.executor.memoryOverhead（默认仅 executor 堆的 10%），堆外溢出导致 Container 被 YARN 杀掉（exit code 143 / 137），湖仓作业跑 Parquet 压缩时尤甚，建议显式调到 15%；其二，Container 粒度过小（如 1GB）会让集群碎片化、调度效率下降，粒度过大则排队变长，用上面的比例表对齐后不要频繁改动。

适用边界

1. 小规模团队需要 DolphinScheduler 吗？

先回答三个问题：每日作业是否少于 20 个？链路是否单链无分叉？是否没有补数与多租户诉求？如果三个都是"是"，那么 crontab + 依赖探测脚本（或单机 Airflow）完全够用，引入 DS 反而背上"ZooKeeper + 元数据库 + 三类服务进程"的运维成本。

出现以下任意两条，就值得上 DS：

• 作业规模超过 50 个/日，人工排 cron 已经排不开错峰；
• 链路分叉多、跨团队依赖多，需要可视化的 DAG 评审与血缘；
• 有明确的补数（历史区间回跑）需求；
• 多团队共用集群，需要多租户隔离与操作审计；
• 要求调度平台自身 HA（crontab 是单点，机器故障即停摆）。

2. 什么时候不适合标签调度？

• 节点数少且同构：10 节点以下的小集群再切标签，每个分组的容量都很薄，一次节点故障就可能击穿标签分组的承载能力；
• 集群常年高利用率（如整体大于 85%）：独占节点意味着牺牲弹性共享，物理隔离的代价超过收益，此时靠队列 + 抢占更划算；
• 标签维度组合过多（机房 × 业务 × 优先级）：标签组合爆炸会让容量碎片化、排障路径变长，标签维度建议不超过两个；
• 已整体上 K8s：统一调度器（Yunikorn / Volcano）+ Namespace/Quota 天然提供更强的隔离与弹性，再保留 YARN 标签属于双轨冗余。