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 中一次任务执行的参数体系由三部分组成参数池作用域说明globalParam整个工作流实例工作流级别参数对所有节点可见优先级最高varPool任务节点之间任务节点执行后产出的变量池作为节点间数据传递的中间介质localParam单个任务节点用户在定义任务时配置的本地参数包含 IN输入与 OUT输出两种方向用户在定义任务时设置的方向为 OUT 的参数会被保存到该任务的localParam中。这一定义位置是整个机制的起点也是本文标题中全局参数名称的由来——虽然参数定义在单个任务上但通过varPool可以在上下游节点之间全局流动。从整体看一条参数的生命周期包含四个阶段定义用户在任务配置中声明 OUT 参数保存在该任务localParam传递Master 在创建下游 taskInstance 时将直接前置节点preTasks的varPool合并后写入taskInstance.varPool随任务下发消费Worker 端将varPool与localParam、globalParam按优先级合并并在节点内容执行前用正则将${变量名}替换为对应值产出与回写SQL / SHELL 节点执行后按规则产出 OUT 参数序列化为 JSON 的varPool传回 MasterMaster 再将 OUT 参数回写到localParam供下游继续使用。二、参数的使用Master 如何合并并传递 varPool2.1 前置节点 varPool 的合并规则当 Master 需要创建当前任务节点对应的 taskInstance 时会先从 DAG 中获取该节点的直接前置节点 preTasks读取每个 preTasks 的varPool类型为ListProperty并将这些 varPool合并为一个 varPool。合并过程中若出现同名变量按以下逻辑决定最终取值若所有同名变量的值都为null则合并后的值为null若有且只有一个值为非null则合并后的值为该非null值若所有同名变量的值都不是null则取产生该 varPool 的 taskInstance 的 endtime结束时间最早的那个值。这一合并逻辑在源码中对应 VarPoolUtils.java 的mergeVarPool(ListListProperty)当只有一个 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解析为MapString, Property格式其中map 的 key 为property.prop即变量名value 为完整的 Property 对象包含prop、direct、type、value四个字段对应 Property.java。在 processor任务执行器处理参数时会将varPool、localParam、globalParam三个参数池合并。当出现参数名重复时按以下优先级执行替换高优先级保留低优先级被替换优先级参数池说明高globalParam工作流全局参数最终覆盖同名参数中varPool上游节点产出的变量池低localParam任务本地参数这一规则决定了即使任务本地配置了某个参数的默认值只要上游节点通过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 节点参数返回的结构为ListMapString, String其中List的元素对应每行数据Map的 key 为列名value 为该列对应的值。匹配规则如下若 SQL 语句只返回一行数据则根据用户在定义任务时定义的 OUT 参数名去匹配列名匹配到则将对应列值作为该参数的值未匹配到则放弃。若 SQL 语句返回多行数据则根据用户定义的类型为LIST的 OUT 参数名去匹配列名将该列所有行的数据转换为ListString作为该参数的值若该 OUT 参数类型不是 LIST则不会赋值未匹配到则放弃。这一逻辑在 SqlParameters.java 的dealOutParam(String result)中实现先通过getListMapByString把结果 JSON 解析为ListMapString, String当sqlResult.size() 1时先以第一行数据的列名初始化sqlResultFormat逐行把同名列的值聚合成ListString再对类型为DataType.LIST的 OUT 参数执行JSONUtils.toJsonString序列化赋值当结果只有一行时则直接将首行对应列值String.valueOf后赋值。3.2 SHELL 节点${setValue(keyvalue)} 约定SHELL 节点执行后processor 返回的结果为MapString, String。用户在编写 Shell 脚本时需要在脚本输出中显式声明以下形式的特殊标记echo ${setValue(keyvalue)}参数处理时会去掉${setValue()}外壳按照进行拆分第 0 段为 key第 1 段为 value。随后同样匹配用户在定义任务时声明的 OUT 参数名与 key将 value 作为该参数的值。上述解析逻辑由 TaskOutputParameterParser.java 完成几个工程细节值得注意同时支持${setValue(...)}与#{setValue(...)}两种写法appendParseLog中依次探测两种前缀拆分时使用split(, 2)即只按第一个拆分value 中可以安全地包含字符单个参数默认最多解析1024 行maxOneParameterRows超过行数或长度上限默认Integer.MAX_VALUE的参数会被跳过并记录 warn 日志这是为了防止日志中未闭合的表达式导致内存溢出OOM支持参数表达式跨多行输出解析器会持续累积日志行直到找到)}结束标记。四、返回参数处理与 varPool 回传 Master4.1 Worker 端返回参数的统一处理流程无论 SQL 还是 SHELL 节点Worker 端对返回参数的处理遵循同一套流程获取 processor 的执行结果String类型判断 processor 结果是否为空为空则直接退出判断localParam是否为空为空则退出获取localParam中方向为 OUT 的参数若为空则退出将结果 String 按上述格式解析SQL 解析为ListMapString, StringSHELL 解析为MapString, String将匹配好值的参数赋值给varPoolListProperty其中保留原有方向为 IN 的参数。注意第 6 步的关键点varPool 中会保留节点原有的 IN 参数。从 AbstractParameters.java 的dealOutParam(MapString, 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_dataSQL与consume_dataSHELL两个串行节点。定义在generate_data的自定义参数中定义 OUT 参数table_count数据类型 VARCHAR并执行SELECT COUNT(*) AS table_count FROM information_schema.tables;。产出SQL 返回单行结果SqlParameters.dealOutParam按列名table_count匹配 OUT 参数并赋值该参数与原有 IN 参数共同写入varPool。回传varPool序列化为 JSON 传回 MasterMaster 将 OUT 参数table_count回写到generate_data的localParam。合并传递创建consume_data的 taskInstance 时Master 读取前置节点generate_data的varPool将table_count的方向更新为 IN 后写入consume_data.varPool并下发 Worker。消费Worker 端将varPool解析为MapString, Property与localParam、globalParam按globalParam varPool localParam优先级合并随后将consume_data脚本中的${table_count}替换为实际数值后再执行。如果此时工作流级globalParam中也定义了table_count则下游实际拿到的将是globalParam的值而非上游 SQL 产出的值——这正是合并优先级规则的实战体现。六、实战注意事项与排查建议结合文档约定与源码实现使用全局参数时建议关注以下几点OUT 参数命名与列名/输出 key 必须完全一致SQL 节点按列名精确匹配、SHELL 节点按拆分后的 key 精确匹配不一致的参数会被静默放弃。多行结果必须配合 LIST 类型SQL 返回多行时只有类型为 LIST 的 OUT 参数才会被赋值普通类型在多行场景下不产生值。同名冲突的取值规则多前置节点产出同名参数时全部为 null 取 null、唯一非 null 取该值、全部非 null 取 endtime 最早节点合并后方向统一变为 IN。优先级陷阱globalParam会覆盖varPool与localParam中的同名参数排查参数值不对时先检查工作流全局参数。SHELL 输出格式务必完整输出${setValue(keyvalue)}外壳也支持#{setValue(...)}写法value 中包含时解析器只按第一个拆分可以安全使用但单个参数的输出行数建议控制在 1024 行以内避免被安全上限截断。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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
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 中一次任务执行的参数体系由三部分组成参数池作用域说明globalParam整个工作流实例工作流级别参数对所有节点可见优先级最高varPool任务节点之间任务节点执行后产出的变量池作为节点间数据传递的中间介质localParam单个任务节点用户在定义任务时配置的本地参数包含 IN输入与 OUT输出两种方向用户在定义任务时设置的方向为 OUT 的参数会被保存到该任务的localParam中。这一定义位置是整个机制的起点也是本文标题中全局参数名称的由来——虽然参数定义在单个任务上但通过varPool可以在上下游节点之间全局流动。从整体看一条参数的生命周期包含四个阶段定义用户在任务配置中声明 OUT 参数保存在该任务localParam传递Master 在创建下游 taskInstance 时将直接前置节点preTasks的varPool合并后写入taskInstance.varPool随任务下发消费Worker 端将varPool与localParam、globalParam按优先级合并并在节点内容执行前用正则将${变量名}替换为对应值产出与回写SQL / SHELL 节点执行后按规则产出 OUT 参数序列化为 JSON 的varPool传回 MasterMaster 再将 OUT 参数回写到localParam供下游继续使用。二、参数的使用Master 如何合并并传递 varPool2.1 前置节点 varPool 的合并规则当 Master 需要创建当前任务节点对应的 taskInstance 时会先从 DAG 中获取该节点的直接前置节点 preTasks读取每个 preTasks 的varPool类型为ListProperty并将这些 varPool合并为一个 varPool。合并过程中若出现同名变量按以下逻辑决定最终取值若所有同名变量的值都为null则合并后的值为null若有且只有一个值为非null则合并后的值为该非null值若所有同名变量的值都不是null则取产生该 varPool 的 taskInstance 的 endtime结束时间最早的那个值。这一合并逻辑在源码中对应 VarPoolUtils.java 的mergeVarPool(ListListProperty)当只有一个 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解析为MapString, Property格式其中map 的 key 为property.prop即变量名value 为完整的 Property 对象包含prop、direct、type、value四个字段对应 Property.java。在 processor任务执行器处理参数时会将varPool、localParam、globalParam三个参数池合并。当出现参数名重复时按以下优先级执行替换高优先级保留低优先级被替换优先级参数池说明高globalParam工作流全局参数最终覆盖同名参数中varPool上游节点产出的变量池低localParam任务本地参数这一规则决定了即使任务本地配置了某个参数的默认值只要上游节点通过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 节点参数返回的结构为ListMapString, String其中List的元素对应每行数据Map的 key 为列名value 为该列对应的值。匹配规则如下若 SQL 语句只返回一行数据则根据用户在定义任务时定义的 OUT 参数名去匹配列名匹配到则将对应列值作为该参数的值未匹配到则放弃。若 SQL 语句返回多行数据则根据用户定义的类型为LIST的 OUT 参数名去匹配列名将该列所有行的数据转换为ListString作为该参数的值若该 OUT 参数类型不是 LIST则不会赋值未匹配到则放弃。这一逻辑在 SqlParameters.java 的dealOutParam(String result)中实现先通过getListMapByString把结果 JSON 解析为ListMapString, String当sqlResult.size() 1时先以第一行数据的列名初始化sqlResultFormat逐行把同名列的值聚合成ListString再对类型为DataType.LIST的 OUT 参数执行JSONUtils.toJsonString序列化赋值当结果只有一行时则直接将首行对应列值String.valueOf后赋值。3.2 SHELL 节点${setValue(keyvalue)} 约定SHELL 节点执行后processor 返回的结果为MapString, String。用户在编写 Shell 脚本时需要在脚本输出中显式声明以下形式的特殊标记echo ${setValue(keyvalue)}参数处理时会去掉${setValue()}外壳按照进行拆分第 0 段为 key第 1 段为 value。随后同样匹配用户在定义任务时声明的 OUT 参数名与 key将 value 作为该参数的值。上述解析逻辑由 TaskOutputParameterParser.java 完成几个工程细节值得注意同时支持${setValue(...)}与#{setValue(...)}两种写法appendParseLog中依次探测两种前缀拆分时使用split(, 2)即只按第一个拆分value 中可以安全地包含字符单个参数默认最多解析1024 行maxOneParameterRows超过行数或长度上限默认Integer.MAX_VALUE的参数会被跳过并记录 warn 日志这是为了防止日志中未闭合的表达式导致内存溢出OOM支持参数表达式跨多行输出解析器会持续累积日志行直到找到)}结束标记。四、返回参数处理与 varPool 回传 Master4.1 Worker 端返回参数的统一处理流程无论 SQL 还是 SHELL 节点Worker 端对返回参数的处理遵循同一套流程获取 processor 的执行结果String类型判断 processor 结果是否为空为空则直接退出判断localParam是否为空为空则退出获取localParam中方向为 OUT 的参数若为空则退出将结果 String 按上述格式解析SQL 解析为ListMapString, StringSHELL 解析为MapString, String将匹配好值的参数赋值给varPoolListProperty其中保留原有方向为 IN 的参数。注意第 6 步的关键点varPool 中会保留节点原有的 IN 参数。从 AbstractParameters.java 的dealOutParam(MapString, 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_dataSQL与consume_dataSHELL两个串行节点。定义在generate_data的自定义参数中定义 OUT 参数table_count数据类型 VARCHAR并执行SELECT COUNT(*) AS table_count FROM information_schema.tables;。产出SQL 返回单行结果SqlParameters.dealOutParam按列名table_count匹配 OUT 参数并赋值该参数与原有 IN 参数共同写入varPool。回传varPool序列化为 JSON 传回 MasterMaster 将 OUT 参数table_count回写到generate_data的localParam。合并传递创建consume_data的 taskInstance 时Master 读取前置节点generate_data的varPool将table_count的方向更新为 IN 后写入consume_data.varPool并下发 Worker。消费Worker 端将varPool解析为MapString, Property与localParam、globalParam按globalParam varPool localParam优先级合并随后将consume_data脚本中的${table_count}替换为实际数值后再执行。如果此时工作流级globalParam中也定义了table_count则下游实际拿到的将是globalParam的值而非上游 SQL 产出的值——这正是合并优先级规则的实战体现。六、实战注意事项与排查建议结合文档约定与源码实现使用全局参数时建议关注以下几点OUT 参数命名与列名/输出 key 必须完全一致SQL 节点按列名精确匹配、SHELL 节点按拆分后的 key 精确匹配不一致的参数会被静默放弃。多行结果必须配合 LIST 类型SQL 返回多行时只有类型为 LIST 的 OUT 参数才会被赋值普通类型在多行场景下不产生值。同名冲突的取值规则多前置节点产出同名参数时全部为 null 取 null、唯一非 null 取该值、全部非 null 取 endtime 最早节点合并后方向统一变为 IN。优先级陷阱globalParam会覆盖varPool与localParam中的同名参数排查参数值不对时先检查工作流全局参数。SHELL 输出格式务必完整输出${setValue(keyvalue)}外壳也支持#{setValue(...)}写法value 中包含时解析器只按第一个拆分可以安全使用但单个参数的输出行数建议控制在 1024 行以内避免被安全上限截断。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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
