Apache DolphinScheduler 全局参数机制深度解析：OUT 参数从定义、传递到回写的完整链路

Apache DolphinScheduler 全局参数机制深度解析：OUT 参数从定义、传递到回写的完整链路

【免费下载链接】dolphinschedulerApache DolphinScheduler is the modern data orchestration platform. Agile to create high performance workflow with low-code项目地址: https://gitcode.com/GitHub_Trending/dol/dolphinscheduler

本文聚焦 Apache DolphinScheduler 中全局参数的底层实现机制：当你在工作流中定义一个方向为 OUT 的参数后，它如何被保存进localParam、如何在 DAG 前置节点之间通过varPool合并与传递、Worker 端如何解析并替换${变量名}占位符，以及 SQL / SHELL 两类任务节点如何产出并回写参数值。读完本文，你将掌握全局参数在 Master 与 Worker 之间的完整流转链路，并理解同名参数冲突时的合并优先级与取值规则，为排查参数不生效、取值异常等实战问题打下基础。

一、全局参数的定位：三种参数池与一条完整链路

在 DolphinScheduler 中，一次任务执行的参数体系由三部分组成：

用户在定义任务时设置的方向为 OUT 的参数，会被保存到该任务的localParam中。这一定义位置是整个机制的起点，也是本文标题中"全局参数"名称的由来——虽然参数定义在单个任务上，但通过varPool可以在上下游节点之间全局流动。

从整体看，一条参数的生命周期包含四个阶段：

1. 定义：用户在任务配置中声明 OUT 参数，保存在该任务localParam；
2. 传递：Master 在创建下游 taskInstance 时，将直接前置节点（preTasks）的varPool合并后写入taskInstance.varPool，随任务下发；
3. 消费：Worker 端将varPool与localParam、globalParam按优先级合并，并在节点内容执行前用正则将${变量名}替换为对应值；
4. 产出与回写：SQL / SHELL 节点执行后按规则产出 OUT 参数，序列化为 JSON 的varPool传回 Master，Master 再将 OUT 参数回写到localParam，供下游继续使用。

二、参数的使用：Master 如何合并并传递 varPool

2.1 前置节点 varPool 的合并规则

当 Master 需要创建当前任务节点对应的 taskInstance 时，会先从 DAG 中获取该节点的直接前置节点 preTasks，读取每个 preTasks 的varPool（类型为List<Property>），并将这些 varPool合并为一个 varPool。合并过程中若出现同名变量，按以下逻辑决定最终取值：

• 若所有同名变量的值都为null，则合并后的值为null；
• 若有且只有一个值为非null，则合并后的值为该非null值；
• 若所有同名变量的值都不是null，则取产生该 varPool 的 taskInstance 的 endtime（结束时间）最早的那个值。

这一合并逻辑在源码中对应 VarPoolUtils.java 的mergeVarPool(List<List<Property>>)：当只有一个 varPool 时直接返回；多个时以Property#getProp()（变量名）为 key 放入 HashMap，后放入的覆盖先放入的，从而实现"后者取最早 endtime 节点"的效果。

合并过程中，所有被合并过来的 Property 的方向都会被更新为 IN。这一点至关重要：上游产出的 OUT 参数，对于当前节点而言属于"输入"，方向变为 IN 后即可在节点内容中被${变量名}方式引用。合并后的结果保存在taskInstance.varPool中，随任务分发给 Worker。

从源码结构看，Master 端通过VarPoolUtils.mergeVarPoolJsonString(String... varPoolJsons)（见 VarPoolUtils.java）处理多个前置节点的 varPool JSON，序列化与反序列化均走JSONUtils，保证跨进程传输的格式统一。

2.2 Worker 端的参数池合并优先级

Worker 收到任务后，首先将taskInstance.varPool解析为Map<String, Property>格式，其中map 的 key 为property.prop，即变量名，value 为完整的 Property 对象（包含prop、direct、type、value四个字段，对应 Property.java）。

在 processor（任务执行器）处理参数时，会将varPool、localParam、globalParam三个参数池合并。当出现参数名重复时，按以下优先级执行替换（高优先级保留，低优先级被替换）：

这一规则决定了：即使任务本地配置了某个参数的默认值，只要上游节点通过varPool传递了同名参数，就会以varPool中的值为准；而工作流级的globalParam则拥有最终决定权。

2.3 占位符替换

参数合并完成后，会在节点内容实际执行之前，利用正则表达式匹配${变量名}并将其替换为对应的值。也就是说，SQL 语句、Shell 脚本中出现的${变量名}占位符是在任务真正运行前被静态替换的，替换完成后的实际内容才交给执行引擎运行。

对于 SQL 节点，参数占位符还会经历一步特殊处理：当某个参数的类型为LIST时，会将其值（JSON 数组）展开为多个?占位符（见 ParameterUtils.java 中 LIST 类型的展开逻辑），并在扩展 Map 中按原类型构造新 Property，保证WHERE column IN (?, ?)这类动态 SQL 的正确性。

三、参数的设置：SQL 与 SHELL 节点的产出方式

目前 DolphinScheduler 中仅支持 SQL 和 SHELL 两种节点类型的参数获取（即 OUT 参数产出）。实现上都是先从localParam中取出方向为 OUT 的参数，再根据不同节点类型的产出格式做对应处理。

3.1 SQL 节点：单行匹配与 LIST 多行匹配

SQL 节点参数返回的结构为List<Map<String, String>>，其中：

• List的元素对应每行数据；
• Map的 key 为列名，value 为该列对应的值。

匹配规则如下：

• 若 SQL 语句只返回一行数据，则根据用户在定义任务时定义的 OUT 参数名去匹配列名，匹配到则将对应列值作为该参数的值；未匹配到则放弃。
• 若 SQL 语句返回多行数据，则根据用户定义的类型为LIST的 OUT 参数名去匹配列名，将该列所有行的数据转换为List<String>作为该参数的值；若该 OUT 参数类型不是 LIST，则不会赋值；未匹配到则放弃。

这一逻辑在 SqlParameters.java 的dealOutParam(String result)中实现：先通过getListMapByString把结果 JSON 解析为List<Map<String, String>>；当sqlResult.size() > 1时，先以第一行数据的列名初始化sqlResultFormat，逐行把同名列的值聚合成List<String>，再对类型为DataType.LIST的 OUT 参数执行JSONUtils.toJsonString序列化赋值；当结果只有一行时，则直接将首行对应列值String.valueOf后赋值。

3.2 SHELL 节点：${setValue(key=value)} 约定

SHELL 节点执行后，processor 返回的结果为Map<String, String>。用户在编写 Shell 脚本时，需要在脚本输出中显式声明以下形式的特殊标记：

echo '${setValue(key=value)}'

参数处理时会去掉${setValue()}外壳，按照=进行拆分：第 0 段为 key，第 1 段为 value。随后同样匹配用户在定义任务时声明的 OUT 参数名与 key，将 value 作为该参数的值。

上述解析逻辑由 TaskOutputParameterParser.java 完成，几个工程细节值得注意：

• 同时支持${setValue(...)}与#{setValue(...)}两种写法（appendParseLog中依次探测两种前缀）；
• 拆分时使用split("=", 2)，即只按第一个=拆分，value 中可以安全地包含=字符；
• 单个参数默认最多解析1024 行（maxOneParameterRows），超过行数或长度上限（默认Integer.MAX_VALUE）的参数会被跳过并记录 warn 日志，这是为了防止日志中未闭合的表达式导致内存溢出（OOM）；
• 支持参数表达式跨多行输出，解析器会持续累积日志行，直到找到)}结束标记。

四、返回参数处理与 varPool 回传 Master

4.1 Worker 端返回参数的统一处理流程

无论 SQL 还是 SHELL 节点，Worker 端对返回参数的处理遵循同一套流程：

1. 获取 processor 的执行结果（String类型）；
2. 判断 processor 结果是否为空，为空则直接退出；
3. 判断localParam是否为空，为空则退出；
4. 获取localParam中方向为 OUT 的参数，若为空则退出；
5. 将结果 String 按上述格式解析（SQL 解析为List<Map<String, String>>，SHELL 解析为Map<String, String>）；
6. 将匹配好值的参数赋值给varPool（List<Property>，其中保留原有方向为 IN 的参数）。

注意第 6 步的关键点：varPool 中会保留节点原有的 IN 参数。从 AbstractParameters.java 的dealOutParam(Map<String, String> taskOutputParams)可以看到：先取出 OUT 参数，用taskOutputParams中匹配到的值进行注入，最后通过VarPoolUtils.mergeVarPool(Lists.newArrayList(varPool, outProperty))将原 varPool（含 IN 参数）与新的 OUT 参数合并，而不是整体替换。

4.2 序列化回传与 OUT 回写

• 合并后的varPool会被格式化为JSON 字符串传递给 Master（对应 VarPoolUtils.java 的serializeVarPool）。
• Master 接收到 varPool 后，会将其中方向为 OUT 的参数回写到该任务的localParam中，完成参数的持久化闭环。

回写之后，该任务的localParam就携带了最终产出的 OUT 值，下游节点在创建 taskInstance 时又可以读取该任务的varPool，从而形成"产出 → 合并 → 传递 → 消费 → 再产出"的循环链路。

五、一次完整的参数流转示例

用一个最常见的"SQL 产出 → SHELL 消费"场景，把上述链路串起来：

场景：工作流中有generate_data（SQL）与consume_data（SHELL）两个串行节点。

1. 定义：在generate_data的"自定义参数"中定义 OUT 参数table_count（数据类型 VARCHAR），并执行SELECT COUNT(*) AS table_count FROM information_schema.tables;。
2. 产出：SQL 返回单行结果，SqlParameters.dealOutParam按列名table_count匹配 OUT 参数并赋值；该参数与原有 IN 参数共同写入varPool。
3. 回传：varPool序列化为 JSON 传回 Master，Master 将 OUT 参数table_count回写到generate_data的localParam。
4. 合并传递：创建consume_data的 taskInstance 时，Master 读取前置节点generate_data的varPool，将table_count的方向更新为 IN 后写入consume_data.varPool并下发 Worker。
5. 消费：Worker 端将varPool解析为Map<String, Property>，与localParam、globalParam按"globalParam > varPool > localParam"优先级合并，随后将consume_data脚本中的${table_count}替换为实际数值后再执行。

如果此时工作流级globalParam中也定义了table_count，则下游实际拿到的将是globalParam的值，而非上游 SQL 产出的值——这正是合并优先级规则的实战体现。

六、实战注意事项与排查建议

结合文档约定与源码实现，使用全局参数时建议关注以下几点：

1. OUT 参数命名与列名/输出 key 必须完全一致：SQL 节点按列名精确匹配、SHELL 节点按=拆分后的 key 精确匹配，不一致的参数会被静默放弃。
2. 多行结果必须配合 LIST 类型：SQL 返回多行时，只有类型为 LIST 的 OUT 参数才会被赋值；普通类型在多行场景下不产生值。
3. 同名冲突的取值规则：多前置节点产出同名参数时，全部为 null 取 null、唯一非 null 取该值、全部非 null 取 endtime 最早节点；合并后方向统一变为 IN。
4. 优先级陷阱：globalParam会覆盖varPool与localParam中的同名参数，排查"参数值不对"时先检查工作流全局参数。
5. SHELL 输出格式：务必完整输出${setValue(key=value)}外壳（也支持#{setValue(...)}写法），value 中包含=时解析器只按第一个=拆分，可以安全使用；但单个参数的输出行数建议控制在 1024 行以内，避免被安全上限截断。
6. varPool 保留 IN 参数：节点产出的 varPool 会保留原 IN 参数，因此下游能同时消费本节点输入与输出的全部变量。

通过理解这一机制，你可以更精确地设计跨节点数据传递的参数模型，并在参数"没传过去""值不对""类型不匹配"等问题出现时，沿着"定义位置 → 合并规则 → 优先级 → 解析格式"四步快速定位根因。

【免费下载链接】dolphinschedulerApache DolphinScheduler is the modern data orchestration platform. Agile to create high performance workflow with low-code项目地址: https://gitcode.com/GitHub_Trending/dol/dolphinscheduler