采集数据

新建HttpFile数据源

1. 进入数据源页面。

   1. 登录DataWorks控制台，切换至目标地域后，单击左侧导航栏的管理中心，在下拉框中选择对应工作空间后单击进入管理中心。

   2. 进入工作空间管理中心页面后，单击左侧导航栏的数据源，进入数据源页面。

2. 在数据源管理页面，单击新增数据源。

3. 在新增数据源对话框中，选择数据源类型为HttpFile。

4. 在创建HttpFile数据源对话框中，配置各项参数。本教程需按示例值填写。

   参数	描述
   数据源名称	输入数据源名称，本教程请填写user_behavior_analysis_httpfile。
   数据源描述	输入DataWorks案例体验专用数据源，在离线同步配置时读取该数据源即可访问平台提供的测试数据，该数据源只支持数据集成场景去读取，其他模块不支持使用。
   适用环境	勾选开发和生产。 说明 开发环境的数据源创建完成后，需要勾选生产，以同样方式创建生产环境的数据源，否则任务生产执行会报错。
   URL域名	输入https://dataworks-workshop-2024.oss-cn-shanghai.aliyuncs.com。

5. 单击指定的数据集成资源组连通状态列的测试连通性，等待界面提示测试完成，连通状态为可连通。

   重要

   需确保至少一个资源组为可连通状态，否则此数据源无法使用向导模式创建同步任务。

6. 单击完成创建。

新建MySQL数据源

1. 在数据源管理页面，单击新增数据源。

2. 在新增数据源对话框中，选择数据源类型为MySQL。

3. 在新增MySQL数据源对话框中，配置各项参数。本教程需按示例值填写。

   参数	描述
   数据源类型	选择连接串模式。
   数据源名称	输入数据源名称，本教程请填写user_behavior_analysis_mysql。
   数据源描述	输入DataWorks案例体验专用数据源，在离线同步配置时读取该数据源即可访问平台提供的测试数据，该数据源只支持数据集成场景去读取，其他模块不支持使用。
   JDBC URL	输入jdbc:mysql://rm-bp1z69dodhh85z9qa.mysql.rds.aliyuncs.com:3306/workshop。 说明 您可以通过公网连接串的形式添加数据源，即通过公网地址访问平台提供的用于教程测试使用的RDS。
   用户名	输入用户名，本教程请填写workshop。
   密码	输入密码，本教程请填写workshop#2017。
   认证选项	无认证。

4. 单击指定的数据集成资源组连通状态列的测试连通性，等待界面提示测试完成，连通状态为可连通。

   数据同步时，一个任务只能使用一种资源组。您需要测试每个资源组的连通性，以保证同步任务使用的数据集成资源组能够与数据源连通，否则将无法正常执行数据同步任务。如果您需要同时测试多种资源组，请选中相应资源组后，单击批量测试连通性。详情请参见配置资源组与网络连通。

5. 单击完成。

新建私有OSS数据源

本次实验将RDS数据源的用户信息和OSS数据源的日志信息同步至私有OSS数据源。

1. 在数据源管理页面，单击新增数据源。

2. 在新增数据源对话框中，选择数据源类型为OSS。

3. 在新增OSS数据源对话框中，配置各项参数。

   参数	描述
   数据源名称	输入数据源的名称，本示例为test_g。
   数据源描述	对数据源进行简单描述。
   Endpoint	输入http://oss-cn-shanghai-internal.aliyuncs.com。
   Bucket	您准备环境时创建的OSS Bucket的名称，示例为dw-emr-demo。
   AccessKey ID	当前登录账号的AccessKey ID，您可以进入安全信息管理页面复制AccessKey ID。
   AccessKey Secret	输入当前登录账号的AccessKey Secret。 重要 AccessKey Secret只在创建时显示，不支持后续再次查看，请妥善保管。如果AccessKey泄露或丢失，请删除并创建新的AccessKey。

4. 单击指定资源组连通状态列的测试连通性，等待界面提示测试完成，连通状态为可连通。

   重要

   需确保至少一个资源组为可连通状态，否则此数据源无法使用向导模式创建同步任务。

5. 单击完成。

新建业务流程

1. 单击左上方的图标，选择全部产品 > 数据开发与运维 > DataStudio（数据开发）。

2. 在数据开发面板，右键单击业务流程，选择新建业务流程。

3. 在新建业务流程对话框中，输入业务名称和描述。

   重要

   业务名称不能超过128个字符，且必须是大小写字母、中文、数字、下划线（_）以及英文句号（.）。

4. 单击新建。

5. 进入业务流程开发面板，鼠标单击虚拟节点并拖拽至右侧的编辑页面。在新建节点对话框中，输入节点名称为workshop_start_emr，单击确认。

   以同样的方式新建两个离线同步节点，节点名称分别为di_ods_raw_log_d_emr和di_ods_user_info_d_emr，用于后续同步RDS用户信息与OSS日志信息。

6. 通过拖拽连线，将workshop_start_emr节点设置为两个离线同步节点的上游节点。

   

配置workstart节点

1. 在数据开发页面，双击相应业务流程下的虚拟节点。打开该节点的编辑页面，单击右侧的调度配置。

2. 在时间属性区域，设置重跑属性为运行成功或失败皆可重跑。

3. 在调度依赖区域，单击使用工作空间根节点，表示设置workshop_start_emr节点的上游节点为工作空间根节点。

   通常命名为工作空间名_root。

   

4. 配置完成后，单击左上方的图标。

配置离线同步节点

1. 同步MySQL数据源的用户信息至自建的OSS。

   1. 在数据开发页面，双击di_ods_user_info_d_emr节点，进入节点配置页面。

   2. 配置同步网络链接。

      完成以下网络与资源配置后，单击下一步，并根据界面提示完成连通性测试。

      参数	描述
      数据来源	数据来源：MySQL 数据源名称：user_behavior_analysis_mysql
      我的资源组	选择已购买的独享集成数据资源组。
      数据去向	数据去向：OSS 数据源名称：选择前文创建的私有OSS数据源，此处示例为test_g。

   3. 配置同步任务。

      参数	描述
      数据来源	表：选择数据源中的ods_user_info_d。 切分键：建议使用主键或有索引的列作为切分键，仅支持类型为整型的字段。此处设置切分键为uid。
      数据去向	文本类型：选择text类型。 文件名（含路径）：根据您自建OSS的目录进行输入，示例为ods_user_info_d/user_${bizdate}/user_${bizdate}.txt。其中ods_user_info_d为您自建的目录名，$bizdate表示获取前一天的日期。 列分隔符：输入列分隔符为|。

   4. 切换脚本模式，补充配置参数。

      单击工具栏中的图标，在已有的脚本中：

      • 手动修改writeSingleObject参数取值："writeSingleObject": true。

      • 手动添加参数"suffix": ".txt"。

        

        说明

        writeSingleObject和suffix参数仅支持脚本模式进行添加。

   5. 配置完成后，单击工具栏中的图标。

2. 同步HttpFile数据源的日志信息至自建的OSS。

   1. 在数据开发页面，双击di_ods_raw_log_d_emr节点，进入节点配置页面。

   2. 配置同步网络链接。

      完成以下网络与资源配置后，单击下一步，并根据界面提示完成连通性测试。

      参数	描述
      数据来源	数据来源：HttpFile 数据源名称：user_behavior_analysis_httpfile
      我的资源组	选择已购买的独享集成数据资源组。
      数据去向	数据去向：OSS 数据源名称：选择前文创建的私有OSS数据源，此处示例为test_g。

   3. 配置同步任务。

      参数	描述
      数据来源	文本类型：选择text类型。 文件路径：/user_log.txt。 列分隔符：输入列分隔符为|。 压缩格式：包括None、Gzip、Bzip2和Zip四种类型，此处选择None。 是否跳过表头：选择No。
      数据去向	文本类型：选择text类型。 文件名（含路径）：根据您自建OSS的目录进行输入，示例为ods_raw_log_d/log_${bizdate}/log_${bizdate}.txt，其中ods_raw_log_d为您自建的目录名，$bizdate表示获取前一天的日期。 列分隔符：输入列分隔符为|。

   4. 配置字段映射。

      重要

      源数据表中只有一列数据，如果有其他映射过来的空列，您需要手动删除。

   5. 单击工具栏中的图标，在已有的脚本中手动修改"writeSingleObject": true，并添加"suffix": ".txt"。

      说明

      writeSingleObject和suffix参数仅支持脚本模式进行添加。

   6. 配置完成后，单击工具栏中的图标。

新建EMR表

新建两张表ods_user_info_d_emr表、ods_raw_log_d_emr表，分别用于查询同步后的原始的MySQL用户数据和OSS日志数据。

1. 在数据开发页面打开新建的业务流程，右键单击EMR，选择新建节点 > EMR Hive。

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

   此处需要新建两个EMR Hive节点（insert_ods_user_info_d_emr和insert_ods_raw_log_d_emr），用于创建对应的两张表，并根据下图，通过拉线配置节点的上下游依赖关系。

   

3. 分别在EMR Hive节点中输入建表语句，并在调度配置中调度资源组选择为独享调度资源组，然后单击保存并运行各建表语句。

   • 新建ods_user_info_d_emr表。

     双击insert_ods_user_info_d_emr节点，在右侧的编辑页面输入下述建表语句。

     

     CREATE EXTERNAL TABLE IF NOT EXISTS ods_user_info_d_emr
     (
         `uid` STRING COMMENT '用户ID',
         `gender` STRING COMMENT '性别',
         `age_range` STRING COMMENT '年龄段',
         `zodiac` STRING COMMENT '星座'
     ) PARTITIONED BY (
       dt STRING
     )
     ROW FORMAT  delimited fields terminated by '|'
     LOCATION 'oss://dw-emr-demo/ods_user_info_d/';
     
     ALTER TABLE ods_user_info_d_emr ADD IF NOT EXISTS PARTITION (dt='${bizdate}')
     LOCATION 'oss://dw-emr-demo/ods_user_info_d/user_${bizdate}/';

     说明

     上述代码中的location为示例路径，与之前配置离线同步任务时的数据去向相同，需要输入您建立的文件路径名称，其中dw-emr-demo是您准备环境时创建的OSS Bucket的名称。

   • 新建ods_raw_log_d_emr表。

     双击insert_ods_raw_log_d_emr节点，在右侧的编辑页面输入下述建表语句。

     --创建OSS日志对应目标表
     CREATE EXTERNAL TABLE IF NOT EXISTS ods_raw_log_d_emr
     (
       `col` STRING
     ) PARTITIONED BY (
       dt STRING
     );
     ALTER TABLE ods_raw_log_d_emr ADD IF NOT EXISTS PARTITION (dt='${bizdate}')
     LOCATION 'oss://dw-emr-demo/ods_raw_log_d/log_${bizdate}/';

     说明

     上述代码中的location为示例路径，与之前配置离线同步任务时的数据去向相同，需要输入您建立的文件路径名称，其中dw-emr-demo是您准备环境时创建的OSS Bucket的名称。

运行并查看结果

运行业务流程

1. 在数据开发页面，双击业务流程下的WorkShop，打开Workflow业务流程面板后，单击工具栏中的图标，按照上下游依赖关系运行业务流程。

2. 确认执行情况。

   • 查看任务运行状态：节点处于状态，即代表同步执行过程无问题。

   • 查看任务执行日志：右键单击insert_ods_user_info_d_emr、insert_ods_raw_log_d_emr节点，选择查看日志。当日志中出现如下字样，表示同步节点运行成功，并成功同步数据。

     

查询同步结果

1. 新建临时查询文件

   在数据开发页面的左侧导航栏，单击，进入临时查询面板。右键单击临时查询，选择新建节点>EMR Hive。

2. 查询同步结果表。

   执行如下SQL语句，确认同步数据写入结果。查看导入ods_raw_log_d_emr和ods_user_info_d_emr的记录数。

   --查询语句中的分区列需要更新为业务日期。例如，任务运行的日期为20191107，则业务日期为20191106，即任务运行日期的前一天。
   SELECT * from ods_user_info_d_emr where dt=业务日期; 
   SELECT * from ods_raw_log_d_emr where dt=业务日期;