最佳实践：数据开发工作流中配置数据推送节点

步骤一：搭建推送流程

双击打开数据推送Demo实例业务流程图页面，依次将查询上月销售总额通过达标条件判断分成达标上月销售达标数据、推送销售达标前三名类别和不达标上月销售不达标数据、推送销售不达标倒数三名类别，关联起来，生成一条如下图所示的条件推送的工作流。

步骤二：配置SQL查询节点

通过配置SQL查询节点对测试数据进行查询，通过节点上下文参数获取到SQL查询节点的输出参数outputs，从而实现将SQL的查询结果传递至分支节点。

1. 双击打开查询上月销售总额节点，编写以下查询SQL代码。

   -- 统计上个月销售额
   SELECT SUM(sales) AS sales_amount
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59');

2. 编辑完成SQL代码后，单击右侧调度配置，在弹出的调度配置面板中配置如下内容。

   • 定时调度时间：08：00。

   • 调度资源组：选择已创建好的调度资源组。

     说明

     初次使用数据推送节点，需先提工单升级调度资源组。

   • 依赖上游节点：勾选使用工作空间根节点。

   • 节点上下文参数：单击本节点输出参数后的添加赋值参数，添加输出参数作为下游节点输入参数的取值。

步骤三：配置分支节点

在分支节点内，通过节点上下文参数的本节点输入参数获取上游SQL查询节点的输出参数outputs，然后再分支节点进行条件判断后通过本节点输出参数输出符合条件的数据。

1. 双击打开达标条件判断节点，单击添加分支，新增两个分支，分别为达标和不达标，详细配置信息如下

   配置项	达标分支	不达标分支
   分支条件	${inputs[0][0]}>=500000。	${inputs[0][0]}<500000。
   关联到节点输出	达标。	不达标。
   分支描述	上月销售总额达标。	上月销售总额不达标。

   说明

   [0][0]为二维数组，用来定位获取到的参数中用于条件判断的数据。

   • 上游为SQL查询节点情况下，使用二位数组，定位参数中需要作为条件的数据。

   • 上游为Python节点情况下，使用一维数组，定位参数中需要作为条件的数据。

2. 双击打开达标条件判断节点，单击右侧调度配置，在弹出的调度配置中配置如下内容。

   参数类型		参数内容与说明
   时间属性	调度周期	日。
   	定时调度时间	08：00。
   	重跑属性	运行成功或失败后皆可重跑。
   资源属性	调度资源组	选择已创建好的调度资源组。 说明 初次使用数据推送节点，需先提工单升级调度资源组。
   本节点输出名称		当配置分支后，输出名中会自动解析并添加到此处。输出名与分支定义的关联到节点输出一致。
   节点上下文参数	本节点输入参数	参数名：inputs。 取值来源：选择上游节点查询上月销售总额的输出参数outputs。
   	本节点输出参数	系统默认添加为outputs。

3. 完成配置后，单击保存达标条件判断节点

步骤四：配置分支SQL查询节点

分支SQL查询节点在本实践中分为上月销售达标数据和上月销售不达标数据两个节点，需要将分支节点输出的达标与不达标节点，配置为分支SQL查询节点的上游节点后，将分支节点SQL查询结果用节点上下文参数的本节点输出参数分别输出至对应的达标与不达标的数据推送节点。

1. 分别双击上月销售达标数据节点和上月销售不达标数据节点，进入编辑页面，编写以下SQL代码

   上月销售达标数据

   SET @all_cat_sales_volume_month := 0.0;
   SELECT SUM(sales) INTO @all_cat_sales_volume_month from orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59');
   
   -- 创建临时表
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
     category VARCHAR(255),
     sales DOUBLE,
     all_cat_sales_volume_month DOUBLE
   );
   --查询并写入临时表
   INSERT INTO temp_array (category, sales, all_cat_sales_volume_month) SELECT category, SUM(sales) as amount, @all_cat_sales_volume_month from orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59') GROUP BY category order by amount desc limit 3;
   --查询达标前三名数据
   SELECT category, sales, all_cat_sales_volume_month from temp_array;

   上月销售不达标数据

   SET @all_cat_sales_volume_month := 0.0;
   SELECT SUM(sales) INTO @all_cat_sales_volume_month from orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59');
   
   --创建临时表
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
     category VARCHAR(255),
     sales DOUBLE,
     all_cat_sales_volume_month DOUBLE
   );
   
   --查询并写入临时表
   INSERT INTO temp_array (category, sales, all_cat_sales_volume_month) SELECT category, SUM(sales) as amount, @all_cat_sales_volume_month from orders WHERE datetime BETWEEN DATE_FORMAT(CURRENT_DATE - INTERVAL 1 MONTH, '%Y-%m-01 00:00:00') AND DATE_FORMAT(LAST_DAY(CURRENT_DATE - INTERVAL 1 MONTH), '%Y-%m-%d 23:59:59') GROUP BY category order by amount asc limit 3;
   
   --查询不达标的后三名数据
   SELECT category, sales, all_cat_sales_volume_month from temp_array;

2. 编辑完成SQL代码后，单击右侧调度配置，在弹出的调度配置面板中配置如下内容。

   • 定时调度时间：08：00。

   • 调度资源组：选择已创建好的调度资源组。

     说明

     初次使用数据推送节点，需先提工单升级调度资源组。

   • 依赖的上游节点：添加上游的分支节点，不同的分支SQL查询节点添加对应的上游输出节点。

     • 上月销售达标数据节点：添加上游节点输出名为达标的节点为上游节点。

     • 上月销售不达标数据节点：添加上游节点输出名为不达标的节点为上游节点。

   • 节点上下文参数：单击本节点输出参数后的添加赋值参数，添加输出参数作为下游节点输入参数的取值。

3. 单击保存上月销售达标数据节点和上月销售不达标数据节点。

步骤五：配置分支数据推送节点

需要创建两个数据推送节点，分别配置节点上下文参数的本节点输入参数，来获取上月销售达标数据和上月销售不达标数据的分支SQL查询节点输出的outputs参数，在正文中使用这些参数并将其推送至各自的目标。

1. 双击推送销售达标前三名类别节点和推送销售不达标倒数三名类别节点，进入节点后，单击调度配置，在调度配置面板中进行如下配置。

   参数类型		参数内容	图示
   调度参数	参数名	curdate
   	参数值	$[yyyymmddhh:mi:ss]
   时间属性	调度周期	日
   	定时调度时间	08：00 说明 确保数据能在08：00时推送至目标渠道。
   	重跑属性	运行成功或失败后皆可重跑。
   资源属性	调度资源组	选择已创建好的调度资源组。 说明 初次使用数据推送节点，需先提工单升级调度资源组。
   节点上下文参数	本节点输入参数	单击添加新增本节点输入参数： 参数名：inputs。 取值来源： 在推送销售达标前三名类别节点选择上月销售达标数据节点的输出参数outputs 在推送销售不达标倒数三名类别节点选择上月销售不达标数据节点的输出参数outputs	推送销售达标前三名类别节点 推送销售不达标倒数三名类别节点

2. 调度配置完成后，即可配置数据推送内容，详情如下。

   • 数据推送目标：下拉数据推送目标选择所需的数据推送目标，若不存在，可单击下拉框右下角的创建数据推送目标，新建推送目标。

     参数	说明
     类型	支持圈选钉钉与飞书。
     对象名称	可按业务需求进行自定义。
     WebHook	钉钉或飞书机器人的Webhook，需要在相应的目标平台上获取。

   • 标题：推送销售达标前三名类别和推送销售不达标倒数三名类别。

   • 正文：按需求进行配置，详情可参见配置推送内容。

     说明

     正文中可直接使用上游SQL查询节点查询的字段名作为占位符进行获取上游输入的参数。

     • 推送销售达标前三名类别示例

     • 推送销售不达标倒数三名类别示例

• 配置完成后，单击保存推送销售达标前三名类别节点和推送销售不达标倒数三名类别节点。

步骤六：测试条件推送流程

在完成条件推送流程配置后，需要对条件测试流程进行测试，以便后续提交发布。

1. 双击打开数据推送Demo实例业务流程页面。

2. 选中查询上月销售总额节点，右键选择运行节点及下游，等待运行完成即可。

   说明

   如果出现任务运行失败，选中需要查看的节点，单击右键选择查看日志即可查看日志。

   

步骤一：搭建推送流程

双击打开数据推送Demo实例业务流程图页面，在业务流程图页面，可以看到已创建的多个节点，依次将查询上一周销售数据、组织销售前三名列表及推送上周前三名类别进行关联，生成一条如下图所示的脚本推送的工作流。

步骤二：配置SQL查询节点

通过配置SQL查询节点对测试数据进行查询，可以通过节点上下文参数的本节点输出参数获取到SQL查询节点的输出参数outputs，从而实现将SQL的查询结果传递至赋值节点。

1. 双击查询上一周销售数据节点，进入编辑页面，编写以下SQL代码。

   -- 统计上一周销售额
   SELECT category, SUM(sales) as amount
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 WEEK), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 23:59:59')
   GROUP BY category ORDER BY amount DESC limit 3;

2. 编辑完成SQL代码后，单击右侧调度配置，在弹出的调度配置面板中配置如下内容。

   • 定时调度时间：08：00。

   • 调度资源组：选择已创建好的调度资源组。

     说明

     初次使用数据推送节点，需先提工单升级调度资源组。

   • 依赖上游节点：勾选使用工作空间根节点。

   • 节点上下文参数：单击本节点输出参数后的添加赋值参数，添加输出参数作为下游节点输入参数的取值。

3. 单击保存查询上一周销售数据节点。

步骤三：配置赋值节点

赋值节点通过节点上下文参数下的本节点输入参数可以获取到上游SQL节点的输出参数outputs，将SQL节点的输出参数进行重新赋值调整，生成新的本节点输出参数，将新数据输出至数据推送节点。

1. 双击组织销售前三名列表的赋值节点，进入节点编辑页面，编写以下Python代码。

   def main():
   
       from datetime import date
       today = date.today()
       formatted_date = today.strftime('%Y-%m-%d')
       
       msg = 'Stat date: ' + formatted_date + ' \\n\\n ' \
       '- 1: ${inputs[0][0]}, sales: ${inputs[0][1]} \\n\\n ' \
       '- 2: ${inputs[1][0]}, sales: ${inputs[1][1]} \\n\\n ' \
       '- 3: ${inputs[2][0]}, sales: ${inputs[2][1]} \\n\\n '
       
       print(msg)
   
   
   if __name__ == "__main__":
       import sys
       main()

2. 编辑完成赋值代码后，单击右侧调度配置，在弹出的调度配置面板中配置如下内容。

   • 定时调度时间：08：00。

   • 调度资源组：选择已创建好的调度资源组。

   • 节点上下文参数。

     • 本节点输入参数：

       • 参数名：inputs。

       • 取值来源：上游查询上一周销售数据节点的输出参数outputs。

     • 本节点输出参数：由系统默认生成。

3. 单击保存组织销售前三名列表赋值节点。

步骤四：配置数据推送节点

通过配置节点上下文参数的本节点输入参数，来获取上游赋值节点输出的outputs参数，在正文中使用这些参数并将其推送至目标。

1. 双击名为推送上周前三名类别的数据推送节点，进入节点后，单击调度配置，在调度配置面板中进行如下配置。

   参数类型		参数内容	图示
   调度参数	参数名	curdate。
   	参数值	$[yyyymmddhh:mi:ss]。
   时间属性	调度周期	日。
   	定时调度时间	08：00。 说明 确保数据能在08：00时推送至目标渠道。
   	重跑属性	运行成功或失败后皆可重跑。
   资源属性	调度资源组	选择已创建好的调度资源组。 说明 初次使用数据推送节点，需先提工单升级调度资源组。
   节点上下文参数	本节点输入参数	单击添加新增本节点输入参数： 参数名：inputs。 取值来源：选择上游节点上周销售前三类的输出参数outputs。

2. 调度配置完成后，即可配置数据推送内容，详情如下。

   • 数据推送目标：下拉数据推送目标选择所需的数据推送目标，若不存在，可单击下拉框右下角的创建数据推送目标，新建推送目标。

     参数	说明
     类型	支持圈选钉钉与飞书。
     对象名称	可按业务需求进行自定义。
     WebHook	钉钉或飞书机器人的Webhook，需要在相应的目标平台上获取。

   • 标题：推送上周前三名类别。

   • 正文：按需求进行配置正文即可，详情可参见配置推送内容。

     说明

     正文中可直接使用上游SQL查询节点查询的字段名作为占位符进行获取上游输入的参数。

     

3. 配置完成后，单击保存推送上周前三名类别推送节点。

步骤五：测试脚本推送流程

在完成脚本推送流程配置后，需要对脚本测试流程进行测试，以便后续提交发布。

1. 双击数据推送Demo实例打开业务流程图页面。

2. 选中查询上一周销售数据节点，右键选择运行节点及下游，等待运行完成即可。

   说明

   如果出现任务运行失败，选中需要查看的节点，单击右键选择查看日志即可查看日志。

   

步骤一：搭建推送流程

双击打开数据推送Demo实例业务流程图页面，在业务流程图页面，可以看到已创建的多个节点，拖拽上游节点查询昨日销售总额、查询昨日销售增长额与下游节点推送昨日销售总额与增长额关联起来，生成一条如下图所示的合并推送的工作流。

步骤二：配置SQL查询节点

通过配置SQL查询节点对测试数据进行查询，可以通过节点上下文参数的本节点输出参数获取到SQL查询节点的输出参数outputs，将SQL的查询结果传递至数据推送节点。

1. 单击查询昨日销售增长额节点，编写以下查询SQL代码。

   -- 统计前日数据
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array1 (
     category VARCHAR(255),
     sales DOUBLE
   );
   -- 写入前日数据
   INSERT INTO temp_array1 (category, sales) SELECT category, SUM(sales)
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 2 DAY), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 2 DAY), '%Y-%m-%d 23:59:59')
   GROUP BY category;
   
   -- 统计昨日数据
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array2 (
     category VARCHAR(255),
     sales DOUBLE
   );
   -- 查询写入昨日数据
   INSERT INTO temp_array2 (category, sales) SELECT category, SUM(sales)
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 23:59:59')
   GROUP BY category;
   
   -- 统计昨日销售增长额
   CREATE TEMPORARY TABLE IF NOT EXISTS result (
     category VARCHAR(255),
     diff DOUBLE
   );
   -- 查询写入增长额数据
   INSERT INTO result (category, diff) select temp_array2.category as category, temp_array2.sales - temp_array1.sales as diff from temp_array1 left join temp_array2 on temp_array1.category = temp_array2.category;
   
   -- 查询增长额临时表
   select category, diff from result;

2. 编辑完成SQL代码后，单击右侧调度配置，在弹出的调度配置面板中配置如下内容。

   • 定时调度时间：08：00。

   • 调度资源组：选择已创建好的调度资源组。

     说明

     初次使用数据推送节点，需先提工单升级调度资源组。

   • 依赖上游节点：勾选使用工作空间根节点。

   • 节点上下文参数：单击本节点输出参数后的添加赋值参数，添加输出参数作为下游节点输入参数的取值。

3. 单击保存查询昨日销售增长额节点。

步骤三：配置数据推送节点

通过节点上下文参数的本节点输入参数接收上游查询昨日销售总额与查询昨日销售增长额的输出参数outputs，从而可以在正文中使用参数并将其推送至目标。

1. 双击名为推送昨日销售总额与增长额的数据推送节点，进入节点后，单击调度配置，在调度配置面板中进行如下配置。

   参数类型		参数内容	图示
   调度参数	参数名	curdate。
   	参数值	$[yyyymmddhh:mi:ss]。
   时间属性	调度周期	日。
   	定时调度时间	08：00。 说明 确保数据能在08：00时推送至目标渠道。
   	重跑属性	运行成功或失败后皆可重跑。
   资源属性	调度资源组	选择已创建好的调度资源组。 说明 初次使用数据推送节点，需先提工单升级调度资源组。
   节点上下文参数	本节点输入参数	单击添加新增本节点输入参数： 参数一： 参数名：inputs1。 取值来源：选择上游节点查询昨日销售总额的输出参数outputs。 参数二： 参数名：inputs2。 取值来源：选择上游节点查询昨日销售增长额的输出参数outputs。

2. 调度配置完成后，即可配置数据推送内容，详情如下。

   • 数据推送目标：下拉数据推送目标选择所需的数据推送目标，若不存在，可单击下拉框右下角的创建数据推送目标，新建推送目标。

     参数	说明
     类型	支持圈选钉钉与飞书。
     对象名称	可按业务需求进行自定义。
     WebHook	钉钉或飞书机器人的Webhook，需要在相应的目标平台上获取。

   • 标题：推送昨日销售总额与增长额

   • 正文：按需求进行配置正文即可，详情可参见配置推送内容。

     说明

     正文中可直接使用上游SQL查询节点查询的字段名作为占位符进行获取上游输入的参数。

     

3. 配置完成后，单击保存推送昨日销售总额与增长额推送节点。

步骤四：测试合并推送流程

在完成简单推送流程配置后，需要对合并推送测试流程进行测试，以方便后续提交发布。

1. 双击打开数据推送Demo实例业务流程图页面。

2. 选中推送昨日销售总额与增长额节点，右键选择运行到该节点，等待运行完成即可。

   说明

   如果出现任务运行失败，选中需要查看的节点，单击右键选择查看日志即可查看日志。

   

步骤一：搭建推送流程

双击数据推送Demo实例后，在业务流程图页面，可以看到已创建的多个节点，需拖动MySQL节点查询昨日销售总额作为上游节点，并将其与作为下游的数据推送节点推送昨日销售总额关联起来，生成一条如下图所示的简单推送的工作流。

步骤二：配置SQL查询节点

通过配置SQL查询节点对测试数据进行查询，可以通过节点上下文参数的本节点输出参数获取到SQL查询节点的输出参数outputs，从而实现将SQL的查询结果传递至数据推送节点。

1. 双击查询昨日销售总额节点，编写以下查询SQL代码。

   -- 创建临时表temp_array
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array (
     total_amount DOUBLE
   );
   
   -- 将查询到的昨日销售总额写入临时表temp_array中
   INSERT INTO temp_array (total_amount) 
   SELECT SUM(sales)
   FROM orders
   WHERE datetime BETWEEN DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 00:00:00') AND DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 DAY), '%Y-%m-%d 23:59:59');
   
   -- 查询临时表temp_array
   select total_amount from temp_array;

2. 编辑完成SQL代码后，单击右侧调度配置，在弹出的调度配置面板中配置如下内容。

   • 定时调度时间：08：00。

   • 调度资源组：选择已创建好的调度资源组。

     说明

     初次使用数据推送节点，需先提工单升级调度资源组。

   • 依赖上游节点：勾选使用工作空间根节点。

   • 节点上下文参数：单击本节点输出参数后的添加赋值参数，添加输出参数作为下游节点输入参数的取值。

3. 单击保存查询昨日销售总额节点。

步骤三：配置数据推送节点

通过配置节点上下文参数的本节点输入参数，来获取上游SQL查询节点输出的outputs参数，在正文中使用这些参数并将其推送至目标。

1. 双击名为推送昨日销售总额的数据推送节点，进入节点后，单击调度配置，在调度配置面板中进行如下配置。

   参数类型		参数内容	图示
   调度参数	参数名	curdate。
   	参数值	$[yyyymmddhh:mi:ss]。
   时间属性。	调度周期	日。
   	定时调度时间	08：00。 说明 确保数据能在08：00时推送至目标渠道。
   	重跑属性	运行成功或失败后皆可重跑。
   资源属性	调度资源组	选择已创建好的调度资源组。 说明 初次使用数据推送节点，需先提工单升级调度资源组。
   节点上下文参数	本节点输入参数	单击添加新增本节点输入参数： 参数名 : inputs。 取值来源：选择上游节点查询昨日销售总额的输出参数outputs。

2. 调度配置完成后，即可配置数据推送内容，详情如下。

   • 数据推送目标：下拉数据推送目标选择所需的数据推送目标，若不存在，可单击下拉框右下角的创建数据推送目标，新建推送目标。

     参数	说明
     类型	支持圈选钉钉与飞书。
     对象名称	可按业务需求进行自定义。
     WebHook	钉钉或飞书机器人的Webhook，需要在相应的目标平台上获取。

   • 标题：推送昨日销售总额。

   • 正文：按需求进行配置正文即可，详情可参见配置推送内容。

     说明

     正文中可直接使用上游SQL查询节点查询的字段名作为占位符进行获取上游输入的参数。

     

3. 配置完成后，单击保存推送昨日销售总额推送节点。

步骤四：测试简单推送流程

在完成简单推送流程配置后，需要简单测试流程进行测试，以便后续提交发布。

1. 双击打开数据推送Demo实例业务流程图页面。

2. 选中查询昨日销售总额节点，右键选择运行节点及下游，等待运行完成即可。

   说明

   如果出现任务运行失败，选中需要查看的节点，单击右键选择查看日志即可查看日志。