本文为您介绍如何通过DataWorks采集日志数据至EMR引擎。

前提条件

开始本文的操作前，请准备好需要使用的环境。详情请参见准备环境。

背景信息

根据本次实验模拟的场景，您需要分别新建OSS数据源、RDS数据源，用于存储数据。同时需要新建私有OSS数据源，用于存储同步后的数据。

新建OSS数据源

  1. 进入数据源管理页面。
    1. 登录DataWorks控制台。
    2. 在左侧导航栏，单击工作空间列表。
    3. 单击相应工作空间后的更多图标，选择工作空间配置。
      如果您已在DataWorks的某个功能模块，请单击右上方的工作空间管理图标，进入工作空间管理页面。
    4. 在左侧导航栏，单击数据源管理。
  2. 在数据源管理页面，单击右上方的新增数据源。
  3. 在新增数据源对话框中，选择数据源类型为OSS。
  4. 在新增OSS数据源对话框中，配置各项参数。此处您可以直接按照示例参数进行填写。
    图片
    参数描述
    数据源名称输入数据源名称，示例为oss_workshop_log。
    数据源描述对数据源进行简单描述。
    Endpoint输入Endpoint，示例为http://oss-cn-shanghai-internal.aliyuncs.com。
    Bucket输入Bucket名称，示例为new-dataworks-workshop。
    AccessKey ID输入访问密钥中的AccessKey ID，示例为LTAI4FvGT3iU4xjKotpUMAjS。
    AccessKey Secret输入访问密钥中的AccessKey Secret，示例为9RSUoRmNxpRC9EhC4m9PjuG7Jzy7px。
    重要 为方便您更好地体验DataWorks服务，本次模拟实验提供了仅有只读权限的AK供您试用，以上AK仅可用于模拟实验过程中读取样例数据，真实数据生产环境请替换为您自己的数据源AK。
  5. 连通性测试通过后，单击完成。

新建RDS数据源

  1. 在数据源管理页面，单击右上方的新增数据源。
  2. 在新增数据源对话框中，选择数据源类型为MySQL。
  3. 在新增MySQL数据源对话框中，配置各项参数。此处您可以直接按照示例参数进行填写。
    图片
    参数描述
    数据源类型选择连接串模式。
    数据源名称输入数据源名称，示例为rds_workshop_log。
    数据源描述对数据源进行简单描述。
    适用环境勾选开发。
    说明 开发环境的数据源创建完成后，需要勾选生产，以同样方式创建生产环境的数据源，否则任务生产执行会报错。
    JDBC URL输入jdbc:mysql://rm-bp1z69dodhh85z9qa.mysql.rds.aliyuncs.com:3306/workshop。
    用户名输入用户名，示例为workshop。
    密码输入密码，示例为workshop#2017。
    认证选项无认证。
  4. 测试资源组连通性。
    本案例选择资源组连通性类型为数据集成，单击右下角更多选项，即可添加并测试公共资源组的数据连通性。您可以通过连接串形式添加数据源并配置公网地址，即通过公网地址访问平台提供的用于教程测试使用的RDS。该操作将产生公网流量费，DataWorks公网流量计费说明，详情请参见公网流量计费说明。
    重要 通过实例模式创建的数据源不支持使用公共数据集成资源组跨账号、跨地域等场景同步数据。建议您在实际开发时选择独享数据集成资源组，以获得更多产品能力，保障数据同步的高效性。新增和使用独享数据集成资源组，详情请参见新增和使用独享数据集成资源组。

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

新建私有OSS数据源

本次实验将EMR引擎的数据存储在OSS数据源中。

  1. 在数据源管理页面，单击右上方的新增数据源。
  2. 在新增数据源对话框中，选择数据源类型为OSS。
  3. 在新增OSS数据源对话框中，配置各项参数。
    图片
    参数描述
    数据源名称输入数据源的名称。
    数据源描述对数据源进行简单描述。
    Endpoint输入http://oss-cn-shanghai-internal.aliyuncs.com。
    Bucket您在环境准备中创建的OSS Bucket的名称，示例为dw-emr-demo。
    AccessKey ID当前登录账号的AccessKey ID，您可以进入安全信息管理页面复制AccessKey ID。
    AccessKey Secret输入当前登录账号的AccessKey Secret。
  4. 连通性测试通过后，单击完成。

新建业务流程

  1. 单击左上方的图标图标，选择全部产品 > 数据开发与运维 > DataStudio（数据开发）。
  2. 在数据开发面板，右键单击业务流程，选择新建业务流程。
  3. 在新建业务流程对话框中，输入业务名称和描述。
    重要 业务名称不能超过128个字符，且必须是大小写字母、中文、数字、下划线（_）以及英文句号（.）。
  4. 单击新建。
  5. 进入业务流程开发面板，鼠标单击虚拟节点并拖拽至右侧的编辑页面。在新建节点对话框中，输入节点名称为workstart，单击提交。
    虚拟节点

    以同样的方式新建两个离线同步节点，节点名称分别为Log2oss和User2oss。

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

配置workstart节点

  1. 在数据开发页面，双击相应业务流程下的虚拟节点。打开该节点的编辑页面，单击右侧的调度配置。
  2. 在调度依赖区域，单击使用工作空间根节点，设置workstart节点的上游节点为工作空间根节点。
    由于新版本给每个节点都设置了输入输出节点，所以需要给workstart节点设置一个输入。此处设置其上游节点为工作空间根节点，通常命名为工作空间名_root。工作空间根节点
  3. 配置完成后，单击左上方的保存图标。

配置离线同步节点

  1. 同步RDS数据源的用户信息至自建的OSS。
    1. 在数据开发页面，双击User2oss节点，进入节点配置页面。
    2. 选择数据来源。
      数据来源
      参数描述
      数据源选择MySQL > rds_workshop_log数据源。
      表选择数据源中的ods_user_info_d。
      数据过滤您将要同步数据的筛选条件，暂时不支持limit关键字过滤。此处可以不填写。
      切分键建议使用主键或有索引的列作为切分键，仅支持类型为整型的字段。此处设置切分键为uid。
    3. 选择数据去向。
      数据去向
      参数描述
      数据源选择前文创建的OSS数据源，此处示例为OSS > dw_emr_demo数据源。
      Object前缀根据您自建OSS的目录进行输入，示例为ods_user_info_d/user_${bizdate}/user_${bizdate}.txt。
      文本类型选择text类型。
      列分隔符输入列分隔符为|。
      编码格式默认为UTF-8格式。
      null值表示null值的字符串，此处可以不填写。
      时间格式时间序列化格式，此处可以不填写。
      前缀冲突此处选择替换原有文件。
    4. 配置字段映射。
      配置字段映射
    5. 配置通道控制，单击工具栏中的保存图标。
      通道控制
    6. 单击工具栏中的转换脚本图标，在已有的脚本中手动添加参数"writeSingleObject": "true"和"suffix": ".txt"。
      脚本
      说明
      • writeSingleObject和suffix参数仅支持脚本模式进行添加。
      • 存储的路径object需要与自建OSS中的目录一致。
    7. 配置完成后，单击工具栏中的保存图标。
  2. 同步OSS数据源的日志信息至自建的OSS。
    1. 在数据开发页面，双击Log2oss节点，进入节点配置页面。
    2. 选择数据来源。
      参数描述
      数据源选择OSS > oss_workshop_log数据源。
      Object前缀输入user_log.txt。
      文本类型选择text类型。
      列分隔符输入列分隔符为|。
      编码格式默认为UTF-8格式。
      null值表示null值的字符串，此处可以不填写。
      压缩格式包括None、Gzip、Bzip2和Zip四种类型，此处选择None。
      是否包含表头默认为No。
    3. 选择数据去向。
      参数描述
      数据源选择前文创建的OSS数据源，此处示例为OSS > dw_emr_demo数据源。
      Object前缀根据您自建OSS的目录进行输入，示例为ods_raw_log_d/user_log_${bizdate}/user_log_${bizdate}.txt。
      文本类型选择text类型。
      列分隔符输入列分隔符为|。
      编码格式默认为UTF-8格式。
      null值表示null值的字符串，此处可以不填写。
      时间格式时间序列化格式，此处可以不填写。
      前缀冲突此处选择替换原有文件。
    4. 配置字段映射。
      重要 源数据表中只有一列数据，此处需要删除其它映射过来的空列。
    5. 配置通道控制，单击工具栏中的保存图标。
    6. 单击工具栏中的转换脚本图标，在已有的脚本中手动添加参数"writeSingleObject": "true"和"suffix": ".txt"。
      说明
      • writeSingleObject和suffix参数仅支持脚本模式进行添加。
      • 存储的路径object需要与自建OSS中的目录一致。
    7. 配置完成后，单击工具栏中的保存图标。

新建表

  1. 在数据开发页面打开新建的业务流程，右键单击EMR，选择新建 > EMR Hive。
  2. 在新建节点对话框中，输入节点名称，单击提交。
    此处需要新建两个EMR Hive节点（ods_user_info_d和ods_raw_log_d），分别新建存储同步过来的OSS日志数据和RDS日志数据的两张表。成功
  3. 分别在EMR Hive节点中，选择EMR引擎并输入建表语句，单击保存并运行各建表语句。
    • 新建ods_user_info_d表。
      双击ods_user_info_d节点，在右侧的编辑页面输入下述建表语句。语句
      CREATE EXTERNAL TABLE IF NOT EXISTS ods_user_info_d
(
    `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 ADD IF NOT EXISTS PARTITION (dt=${bizdate})
LOCATION 'oss://dw-emr-demo/ods_user_info_d/user_${bizdate}/';
      说明 上述代码中的location为示例路径，需要输入您建立的文件夹的路径名称。
    • 新建ods_raw_log_d表。
      双击ods_raw_log_d节点，在右侧的编辑页面输入下述建表语句。 
      --创建OSS日志对应目标表
CREATE EXTERNAL TABLE IF NOT EXISTS ods_raw_log_d
(
  `col` STRING
) PARTITIONED BY (
  dt STRING
);
ALTER TABLE ods_raw_log_d ADD IF NOT EXISTS PARTITION (dt=${bizdate})
LOCATION 'oss://dw-emr-demo/ods_raw_log_d/user_log_${bizdate}/';
      说明 上述代码中的location为示例路径，需要输入您建立的文件夹的路径名称。
  4. 查看数据同步结果。
    建表语句运行成功后，分别在两个EMR Hive节点中输入查询语句。
    说明 查询语句中的分区列需要更新为业务日期。例如，任务运行的日期为20191107，则业务日期为20191106，即任务运行日期的前一天。
    • 查询ods_user_info_d表的数据。 
      SELECT * from ods_user_info_d where dt=业务日期; --业务日期为任务运行日期的前一天。
      查询
    • 查询ods_raw_log_d表的数据。 
      SELECT * from ods_raw_log_d where dt=业务日期; --业务日期为任务运行日期的前一天。
      查询

后续步骤

现在，您已经学习了如何进行日志数据同步，完成数据的采集，您可以继续下一个教程。在该教程中，您将学习如何对采集的数据进行计算与分析。详情请参见加工数据。