调度变量

步骤一：定义调度变量

1. 进入数据开发页面，双击目标工作流，进入工作流编辑画布。

2. 单击右侧调度配置，找到调度变量区域。

3. 单击添加变量，填写变量信息。

   配置项	说明
   变量名	变量的唯一标识，在工作流内不可重复。建议仅使用字母、数字、下划线。
   变量值	变量的初始值。 任务每次启动运行时，变量从该值开始生效。支持填写常量、工作空间参数与 $[yyyy-mm-dd] 等时间表达式，详情可参考：调度参数来源及其表达式。 不支持在取值中引用工作流参数（如 ${workflow.参数名}）或赋值节点的输出值。

   例如：env=prod、ds_flag=$[yyyymmdd]。

4. 保存工作流。

步骤二：在节点代码中引用变量

在工作流内节点的代码中，使用 ${workflow.variable.变量名} 格式引用。变量名支持字母、数字、下划线与中划线。

以SHELL节点为例，在工作流上配置了env和ds_flag，在代码编辑器中填写如下代码，并保存。

#!/bin/bash
echo "env=${workflow.variable.env}"
echo "ds_flag=${workflow.variable.ds_flag}"

任务运行时，系统在节点执行前将占位符替换为变量在本次运行中的当前值（初始值或本次运行改写后的值），再交给引擎执行。

• 调度变量与调度参数可以同时使用：${workflow.variable.env} 按调度变量规则替换，${bizdate} 仍按调度参数规则替换为业务日期。

• 未知变量名不会报错替换失败：占位符原样保留在代码中，请确保引用的变量已在工作流上定义。

• 调度变量只能在代码区引用。在调度配置的脚本参数中，参数值里书写 ${workflow.variable.变量名} 不会取到变量值。

• 不支持取变量值中的一部分，${workflow.variable.变量名[0][1]}、${workflow.variable.变量名.name} 等写法无效。需要拆分取值时，在变量赋值节点中用 Shell 或 Python3 计算后写入另一个变量。

步骤三：测试运行

1. 在工作流编辑画布，单击页面上方的运行。

2. 在填写运行参数表单的工作流变量列，按需修改本次运行的取值。此处的赋值为调试运行时临时赋值，仅在本次运行时生效，也不会改变变量的初始赋值。

3. 执行运行，并在节点运行日志中检查变量替换结果是否符合预期。

步骤四：使用变量赋值节点在运行期修改变量

需要在运行过程中修改变量取值时，向工作流中添加变量赋值节点。该节点属于通用节点，支持 Shell 与 Python3，通过在脚本中向标准输出打印赋值标记写入新值。生效规则如下。

1. 每次工作流运行时，系统会为本次运行生成一份调度变量副本。工作流内节点执行时，代码中的 ${workflow.variable.变量名} 从该副本取值替换。

2. 变量赋值节点执行前，其脚本中的 ${workflow.variable.变量名} 也会被替换为变量当前值，因此脚本内可以先读取变量再计算新值。

3. 脚本执行时，通过向标准输出打印赋值标记写入新值。赋值标记支持两种格式：

   • 单行格式 ：##WORKFLOW_VARIABLE::变量名=值；

   • 多行格式： ##WORKFLOW_VARIABLE_BEGIN::变量名 起始、以 ##WORKFLOW_VARIABLE_END::变量名 结束，适用于取值包含换行的场景，如多行数据。

   # 读取变量 env，计算后改写变量 ds_flag
   echo "current env: ${workflow.variable.env}"
   echo "##WORKFLOW_VARIABLE::ds_flag=20260903"
   # 多行赋值
   echo "##WORKFLOW_VARIABLE_BEGIN::result_json"
   echo '{"name":"shirt","count":50}'
   echo '{"name":"coat","count":120}'
   echo "##WORKFLOW_VARIABLE_END::result_json"

4. 节点执行成功后，系统解析标准输出中的赋值标记，将新值写回本次运行的变量副本。该节点之后运行的节点，引用 ${workflow.variable.变量名} 得到的即为新值。

步骤五：发布与验证

1. 完成测试后，提交并发布工作流至生产环境。调度变量随工作流一并发布。

2. 发布成功后，进入运维中心，确认周期任务中变量替换结果符合预期。

步骤一：定义变量

在工作流调度配置中添加 7 个变量。

步骤二：搭建节点链路

按下表创建 6 个节点并连接依赖：set_ds_flag 分三支，summary_report 汇合三支。

三个变量赋值节点的代码如下。

1. 变量赋值节点set_ds_flag，设置变量赋值节点的语言类型为SHELL。

   # set_ds_flag（变量赋值节点，Shell）
   #!/bin/bash
   echo "env=${workflow.variable.env} tbl=${workflow.variable.tbl_prefix} ds_flag(before)=${workflow.variable.ds_flag}"
   echo "##WORKFLOW_VARIABLE::ds_flag=20260903"
   echo "##WORKFLOW_VARIABLE_BEGIN::stats_json"
   echo "{\"table\":\"${workflow.variable.tbl_prefix}user_trade_${workflow.variable.env}\",\"rows\":120,\"ok\":true}"
   echo "##WORKFLOW_VARIABLE_END::stats_json"

2. 变量赋值节点assign_next_day，设置节点语言类型为python3。

   # assign_next_day（变量赋值节点，Python3）
   print("[b_py] ds_flag(a)=" + "${workflow.variable.ds_flag}")
   stats = """${workflow.variable.stats_json}"""
   print("[b_py] stats_json=" + stats)
   print("##WORKFLOW_VARIABLE::ds_flag=20260904")
   print("##WORKFLOW_VARIABLE::chain_flag=py-touched")

3. 变量赋值节点assign_alert_msg，设置节点语言类型为python3。

   # assign_alert_msg（变量赋值节点，Python3）
   msg_table = "${workflow.variable.tbl_prefix}" + "user_trade_" + "${workflow.variable.env}"
   print("[alert_assign] build alert for " + msg_table)
   print("##WORKFLOW_VARIABLE_BEGIN::alert_msg")
   print("[DAILY-ALERT] table: " + msg_table)
   print("[DAILY-ALERT] ds: ${workflow.variable.ds_flag}")
   print("please contact: dw-group-01")
   print("##WORKFLOW_VARIABLE_END::alert_msg")

三个消费节点的代码如下。

1. python节点quality_check。

   # quality_check（Python 节点）
   rows = __import__("json").loads("""${workflow.variable.stats_json}""").get("rows", 0)
   ts = int("${workflow.variable.quality_threshold}")
   print("[quality] rows=%s threshold=%s -> %s" % (rows, ts, "PASS" if rows >= ts else "WARN"))

2. python节点prepare_report。

   # prepare_report（Python 节点）
   print("[prepare] ds_flag=" + "${workflow.variable.ds_flag}" + " chain=" + "${workflow.variable.chain_flag}")
   stats = """${workflow.variable.stats_json}"""
   print("[prepare] report table rows from stats_json=" + stats)

3. python节点summary_report。

   # summary_report（Python 节点）
   print("[summary] ds_flag(final)=" + "${workflow.variable.ds_flag}")
   print("[summary] chain=" + "${workflow.variable.chain_flag}")
   stats = """${workflow.variable.stats_json}"""
   print("[summary] stats_json=" + stats)
   msg = """${workflow.variable.alert_msg}"""
   print("[summary] alert_msg=" + msg)

步骤三：运行并验证结果

返回工作流画布，点击工具栏的运行按钮。运行工作流后，各节点日志中的关键输出如下，变量值随节点运行顺序依次传递。

三种传递场景

配置步骤

1. 在父工作流的编辑画布中，添加SUBPROCESS节点。

2. 在工作流变量区块找到需要传值的子工作流变量，后续也可以单击画布中节点右上角的编辑图标再次修改。

3. 在重置值中填入父工作流变量的引用值，例如 ${workflow.variable.env}。留空表示不重置，子工作流沿用自己的初始值。

4. 保存并发布父工作流。子工作流的变量定义有变动时，子工作流也需重新发布。