大数据开发治理平台 DataWorks：最佳实践：数据开发工作流中配置数据推送节点

步骤一：搭建推送流程

双击打开数据推送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。

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

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

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

步骤三：配置分支节点

在分支节点内，通过节点上下文参数的本节点输入参数获取上游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查询结果用节点上下文参数的本节点输出参数分别输出至对应的达标与不达标的数据推送节点。

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。

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

   • 依赖的上游节点：添加上游的分支节点，因在步骤一：搭建推送流程已创建好流程，需查看上游依赖是否正确。

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

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

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

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

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

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

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

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

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

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

     参数	说明
     类型	支持圈选钉钉、飞书、企业微信与Teams。
     对象名称	可按业务需求进行自定义。
     WebHook	钉钉、飞书、企业微信机器人或Teams的Webhook，需要在相应的目标平台上获取。 说明 飞书机器人Webhook获取方式可参见：飞书Webhook触发器配置。 teams Webhook获取方式可参见：创建传入Webhook。

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

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

     说明

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

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

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

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

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

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

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。

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

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

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

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。 说明 本实践以8:00为例，确保数据能在08:00时推送至目标渠道。您可以根据实际需要配置其他时间点。
   	重跑属性	运行成功或失败后皆可重跑。
   资源属性	调度资源组	选择已创建好的调度资源组。 说明 初次使用数据推送节点，需先提工单升级调度资源组。
   节点上下文参数	本节点输入参数	单击添加新增本节点输入参数： 参数名：inputs。 取值来源：选择上游节点上周销售前三类的输出参数outputs。

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

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

     参数	说明
     类型	支持圈选钉钉、飞书、企业微信与Teams。
     对象名称	可按业务需求进行自定义。
     WebHook	钉钉、飞书、企业微信机器人或Teams的Webhook，需要在相应的目标平台上获取。 说明 飞书机器人Webhook获取方式可参见：飞书Webhook触发器配置。 teams 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。

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

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

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

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

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

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

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

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

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

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

     参数	说明
     类型	支持圈选钉钉、飞书、企业微信与Teams。
     对象名称	可按业务需求进行自定义。
     WebHook	钉钉、飞书、企业微信机器人或Teams的Webhook，需要在相应的目标平台上获取。 说明 飞书机器人Webhook获取方式可参见：飞书Webhook触发器配置。 teams Webhook获取方式可参见：创建传入Webhook。

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

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

     说明

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

     

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

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

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

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

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

   说明

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

   

步骤一：搭建推送流程

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

步骤二：配置SQL查询节点

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

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

   -- 创建统计前日数据表temp_array1
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array1 (
     category VARCHAR(255),
     sales DOUBLE
   );
   -- 写入前日数据至temp_array1
   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;
   
   -- 创建统计昨日数据表temp_array2
   CREATE TEMPORARY TABLE IF NOT EXISTS temp_array2 (
     category VARCHAR(255),
     sales DOUBLE
   );
   -- 写入昨日数据至temp_array2
   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;
   
   -- 创建统计昨日销售增长额result
   CREATE TEMPORARY TABLE IF NOT EXISTS result (
     category VARCHAR(255),
     diff DOUBLE
   );
   -- 写入增长额数据至result
   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;
   
   -- 查询增长额临时表result
   SELECT category, diff FROM result;

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

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

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

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

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

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

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

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

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

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

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

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

     参数	说明
     类型	支持圈选钉钉、飞书、企业微信与Teams。
     对象名称	可按业务需求进行自定义。
     WebHook	钉钉、飞书、企业微信机器人或Teams的Webhook，需要在相应的目标平台上获取。 说明 飞书机器人Webhook获取方式可参见：飞书Webhook触发器配置。 teams Webhook获取方式可参见：创建传入Webhook。

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

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

     说明

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

     

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

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

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

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

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

   说明

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

   

步骤一：搭建推送流程

双击数据推送Demo流程后，在业务流程图页面，可以看到已创建的多个节点，需拖动离线同步节点作为赋值节点的上游。将MySQL同步数据至ODPS、ODPS数据查询以及ODPS数据推送三个节点依次连接起来。

步骤二：配置离线同步节点

通过配置MySQL同步数据至ODPS节点，实现将准备数据阶段在MySQL中写入的测试数据同步至ODPS数据源，以供后续使用。

1. 双击MySQL同步数据至ODPS节点进入离线同步节点的配置。

   配置项		配置内容	图示
   数据来源侧	数据来源	MySQL
   	数据源名称	选择您创建的MySQL数据源。
   我的资源组		选择Serverless 资源组。
   数据去向侧	数据去向	MaxCompute（ODPS）
   	数据源名称	选择该空间绑定的MaxCompute数据源。

   配置完成后，系统会自动进行数据源与资源组之间的连动性测试，待连通性展示为成功后即可单击下一步，进行数据来源与去向的详细配置。

2. 配置数据来源与去向。

   配置项		配置内容	图示
   数据来源	数据源	保持默认： MySQL。 选择您创建的MySQL数据源。
   	表	选择orders表。
   	数据过滤	可按需求进行配置，本实践中置空即可。
   	切分键	您可以将源数据表中某一列作为切分键，建议使用主键或有索引的列作为切分键。
   	数据预览	预览从MySQL数据源中获取到的数据，可用于判断是否符合预期。
   数据去向	数据源	保持默认： MaxCompute（ODPS）。 选择该空间绑定的MaxCompute数据源。
   	Tunnel资源组	即 Tunnel Quota，默认选择“公共传输资源”，即MC的免费quota。
   	表	单击一键生成目标表结构，生成目标表。
   	分区信息	如果您每日增量数据限定在对应日期的分区中，可以使用分区做每日增量，比如配置分区pt值为${bizdate}
   	写入模式	写入前清理已有数据（Insert Overwrite）

3. 配置好来源与去向后，将字段进行同名映射即可。

4. 通道配置。

   • 任务期望最大并发数：由于资源原因或者任务本身特性等原因，实际执行时并发数可能小于等于此值，收费按照实际执行的并发数收费，本实践选择2个并发即可。

   • 同步速率：通过限流可以保护数据来源端或者数据去向端的读写压力，不限流的情况下则会提供现有硬件环境下最大的传输性能，本实践不进行限流。

   • 脏数据策略：不容忍脏数据。

   • 分布式处理能力：默认为关闭状态，需要并发数大于等于8才能开启分布式。

   

5. 调度配置。

   完成集成任务配置后，单击右侧调度配置，在弹出的调度面板中配置如下内容。

   • 调度参数：

     • 参数名：bizdate

     • 参数值：$[yyyymmdd-1]

   • 定时调度时间：08：00

   • 重跑属性：下拉选择运行成功或失败后皆可重跑。

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

   • 调度依赖：勾选使用工作空间根节点，作为离线同步任务的上游节点。

6. 单击保存MySQL同步数据至ODPS离线同步节点。

步骤三：配置赋值节点

ODPS数据源不支持使用ODPS SQL节点进行查询并通过节点上下文参数输出数据至数据推送节点。需采用赋值节点对ODPS数据进行查询处理，再通过节点上下文参数进行输出至数据推送节点。

1. 双击ODPS数据查询节点，进入赋值节点。

   1. 在赋值节点编辑页面上方的请选择赋值语言的下拉框里选择ODPS SQL。

   2. 写入ODPS SQL。

      -- 首先，我们创建一个带有排名的子查询，用于计算每个分区（pt）内按销售额（sales）降序排列的顺序。
      -- DENSE_RANK() 函数会为每个分区内的行提供一个唯一的排名，相等的销售额会得到相同的排名，但不会跳过任何一个排名数字。
      --
      -- 子查询部分：
      -- 1. 从'orders'表中选择需要的列：order_id（订单ID）、category（类别）、sales（销售额）、datetime（时间）和pt（业务日期）。
      -- 2. 使用DENSE_RANK() OVER (PARTITION BY pt ORDER BY sales DESC)：
         -- PARTITION BY pt 指定按照pt字段进行分区，即对每个不同的pt值分别进行排名。
         -- ORDER BY sales DESC 表示在每个分区内按照sales字段的降序来排序，即销售额最高的排在前面。
         -- rank 列将会存储每行在各自分区内的销售排名。
      --
      -- 主查询部分：
      -- 从上述带有排名的子查询中筛选出排名在前三名（rank <= 3）的记录。
      -- 这意味着对于每个业务日期，我们只选取销售额最高的前三个订单的信息。
      
      SELECT
        order_id,          -- 选择订单ID
        category,          -- 选择商品类别
        sales,             -- 选择销售额
        datetime,          -- 选择订单时间
        pt                  -- 选择业务日期
      FROM (
          SELECT
            order_id,
            category,
            sales,
            datetime,
            pt,
            DENSE_RANK() OVER (PARTITION BY pt ORDER BY sales DESC) AS rank  -- 计算每个pt分区内的销售额排名
          FROM orders
          WHERE pt = '${bizdate}'  -- 筛选业务日期为${bizdate}的记录
      ) AS ranked_orders
      WHERE rank <= 3  -- 只保留每个pt分区销售额排名前三的记录

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

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

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

   • 依赖上游节点：检查依赖的上游是否为MySQL同步数据至ODPS的离线同步节点。

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

     

3. 单击保存ODPS数据查询节点。

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

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

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

   参数类型		参数内容	图示
   调度参数	参数名	curdate
   	参数值	$[yyyymmddhh:mi:ss]
   时间属性	调度周期	日
   	定时调度时间	08：00 说明 本实践以8:00为例，确保数据能在08:00时推送至目标渠道。您可以根据实际需要配置其他时间点。
   	重跑属性	运行成功或失败后皆可重跑。
   资源属性	调度资源组	选择数据推送节点功能上线日期后创建的资源组，若资源组为发布日期前，需提工单升级调度资源组。 说明 数据推送节点发布日期为2024年6月28日，DataWorks更多发布记录可参见：功能发布记录。
   节点上下文参数	本节点输入参数	单击添加新增本节点输入参数： 参数名：inputs 取值来源：选择上游节点ODPS数据查询的输出参数outputs。

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

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

     参数	说明
     类型	支持圈选钉钉、飞书、企业微信与Teams。
     对象名称	可按业务需求进行自定义。
     WebHook	钉钉、飞书、企业微信机器人或Teams的Webhook，需要在相应的目标平台上获取。 说明 飞书机器人Webhook获取方式可参见：飞书Webhook触发器配置。 teams Webhook获取方式可参见：创建传入Webhook。

   • 标题：ODPS数据推送

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

     说明

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

3. 配置完成后，单击保存ODPS数据推送节点。

步骤五：测试MaxCompute数据推送流程

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

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

2. 选中MySQL同步数据至ODPS节点，右键选择运行节点及下游，等待运行完成即可。

   说明

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