将分区数据批量同步到以分区名为特征的MySQL表中

步骤一：创建赋值节点

1. 进入数据开发页面，在下拉框中选择对应工作空间后单击进入数据开发。

2. 新建赋值节点。

   1. 右键单击目标业务流程，选择新建节点 > 通用 > 赋值节点。

   2. 在新建节点对话框中，输入名称，并选择节点类型及路径。单击确认，进入赋值节点编辑页面。

3. 配置调度信息。

   单击赋值节点编辑页面右侧的调度配置，进入调度配置页面。

   1. 在调度参数的参数名中输入dt_time，在参数值内点击下拉选择今天$[yyyymmdd]，将当前时间年月日赋值给参数dt_time。

   2. 在调度依赖的依赖的上游节点对话框内点击勾选使用工作空间根节点为节点配置上游依赖。

4. 开发赋值节点任务。

   赋值节点支持ODPS SQL、SHELL和Python三种语言编写开发任务，本文采用ODPS SQL语言进行示例说明。

   SELECT region FROM  sales_data WHERE dt = ${dt_time} GROUP BY region;

   说明

   • 使用动态参数dt_time获取每日年月日信息，可完成对每日全部地域分区的获取。

   • 该节点输出参数默认为右侧调度配置中的本节点输出参数outputs，节点输出值会自动被DataWorks捕获，传递给下游for-each节点。

   • 如果您配置了多个MaxCompute数据源，则需在节点上方的MaxCompute引擎实例对话框中选择所需执行的数据源实例。

5. 保存提交节点任务。

   1. 单击工具栏中的图标，保存编写的SQL语句。

   2. 单击工具栏中的图标，提交节点任务。

   说明

   如您创建的工作空间为标准模式，则需单击节点上方发布按钮进行任务发布。

步骤二：创建for-each节点

1. 新建for-each节点。

   1. 右键单击目标业务流程，选择新建节点 > 通用 > for-each。

   2. 在新建节点对话框中，输入名称，并选择节点类型及路径。单击确认，进入for-each节点编辑页面。

2. 配置调度信息。

   单击赋值节点编辑页面右侧的调度配置，进入调度配置页面。

   1. 在调度依赖的依赖的上游节点下拉选择节点名称，在输入框中输入步骤一中创建的赋值节点名称，选择后缀名为_out的赋值节点，单击添加按钮。

   2. 单击打开节点上下文参数内容，在本节点输入参数中找到参数名为loopDataArray的内容，单击内容右侧编辑按钮。单击选择取值来源信息并进行保存。

步骤三：创建离线同步节点

1. 新建离线同步节点。

   1. 单击for-each节点编辑页面左侧的数据集成 > 离线同步节点。

   2. 在新建节点对话框中，输入节点名称，单击新建。

2. 网络与资源配置。

   选择离线同步节点，右键单击打开节点，进入网络与资源配置页面。

   1. 配置来源数据源，数据来源选择MaxCompute(ODPS)，在数据源名称下拉选择您所需同步的MaxCompute数据源。

   2. 在新建独享数据集成资源组模块选择您所创建的资源组。

   3. 配置去向数据源，数据去向选择MySQL，在数据源名称下拉选择您所需接收的MySQL数据源。

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

   点击下一步进入配置数据来源与去向页面。

   1. 在数据来源中配置schema和表名信息，在schema下拉框选择您所创建的schema，在表名下拉框选择示例表名sales_data。

   2. 在数据去向中配置表名信息，在表名下拉框选择示例表名prefix_20240913_beijing。

4. 配置动态分区参数。

   找到配置数据来源与去向页面上方的转换脚本按钮，单击提示框的确定按钮。

   1. 找到参数partition，将里面的参数值dt=${bizdate},region=信息更换为动态分区dt=${offline_time},region=${dag.foreach.current}。

      说明

      在for-each节点内，可以通过参数${dag.foreach.current}获取当前遍历值。

   2. 找到MySQL表名，将里面的表名prefix_20240912_shanghai信息更换为动态表名prefix_${offline_time}_${dag.foreach.current}。

   3. 在右侧的调度配置对话框中，设置调度参数offline_time，在其参数值内通过下拉菜单选择今天$[yyyymmdd]，从而将当前日期的年月日赋值给参数offline_time。

   4. 在调度依赖的依赖的上游节点的节点输出对话框中输入out，选择节点名称：start的节点进行添加。

5. 单击工具栏中的图标，保存离线同步任务。

步骤四：保存提交for-each循环流程

进入步骤二中创建的for-each节点页面。

1. 如下图所示，配置连接for-each节点与离线同步节点。

   

2. 单击工具栏中的图标，保存循环流程。

3. 单击工具栏中的图标，在提交页面勾选全部节点信息并点击确定按钮进行提交。

步骤五：执行节点任务

1. 单击for-each节点页面左上角的运维，进入运维中心，并在左侧任务运维导航栏中选择周期任务运维 > 周期任务。

2. 找到您所创建的赋值节点，单击补数据 > 当前节点及下游节点进入补数据页面。勾选选择下游任务的任务信息，点击提交并跳转按钮执行该任务。

   说明

   您可以在跳转页面查看对应任务的运行状况。

MaxCompute表数据：

您可根据以下示例SQL语句查询出相应结果。

SELECT * FROM sales_data where dt = '20240913';

MySQL表数据：

您可根据以下示例SQL语句查询出相应结果。

SELECT * FROM `prefix_20240913_beijing`;
SELECT * FROM `prefix_20240913_shanghai`;
SELECT * FROM `prefix_20240913_hangzhou`;

MaxCompute示例数据

-- 创建分区表
-- dt为一级年月日分区,region为二级地域分区。
CREATE TABLE IF NOT EXISTS sales_data (
    id BIGINT,
    product_name STRING,
    quantity INT,
    price DECIMAL(10,2)
)
PARTITIONED BY (dt STRING, region STRING);

-- 插入测试数据
INSERT INTO TABLE sales_data PARTITION (dt='20240913', region='beijing')
VALUES (1, 'phone', 10, 99.99);

INSERT INTO TABLE sales_data PARTITION (dt='20240913', region='shanghai')
VALUES (2, 'book', 5, 70.00);

INSERT INTO TABLE sales_data PARTITION (dt='20240913', region='hangzhou')
VALUES (3, 'book', 5, 70.00);

MySQL建表语句

-- MySql 建表
-- 表名格式prefix_年月日_地域
CREATE TABLE IF NOT EXISTS prefix_20240913_beijing (
    id BIGINT,
    product_name varchar(100),
    quantity INT,
    price DECIMAL(10,2)
);

CREATE TABLE IF NOT EXISTS prefix_20240913_shanghai (
    id BIGINT,
    product_name varchar(100),
    quantity INT,
    price DECIMAL(10,2)
);

CREATE TABLE IF NOT EXISTS prefix_20240913_hangzhou (
    id BIGINT,
    product_name varchar(100),
    quantity INT,
    price DECIMAL(10,2)
);