本文为您介绍如何通过DataWorks采集日志数据至EMR引擎。
前提条件
开始本文的操作前,请准备好需要使用的环境。详情请参见准备环境。
背景信息
根据本次实验模拟的场景,您需要分别新建OSS数据源、RDS数据源,分别用于访问平台提供的测试数据(用户访问网站记录数据、用户信息数据)。同时需要新建私有OSS数据源,用于存储同步后的数据。
新建OSS数据源
进入数据源页面。
登录DataWorks控制台,单击左侧导航栏的管理中心,在下拉框中选择对应工作空间后单击进入管理中心。
进入工作空间管理中心页面后,单击左侧导航栏的数据源,进入数据源页面。
在数据源管理页面,单击新增数据源。
在新增数据源对话框中,选择数据源类型为OSS。
在新增OSS数据源对话框中,配置各项参数。本教程需按示例值填写。
参数
描述
数据源名称
输入数据源名称,本教程请填写oss_workshop_log。
数据源描述
对数据源进行简单描述。
适用环境
如果创建的工作空间为标准模式,勾选开发、生产。开发环境的数据源创建完成后,以同样方式创建生产环境的数据源,否则任务在生产环境执行会报错。
Endpoint
输入Endpoint,本教程请填写
http://oss-cn-shanghai-internal.aliyuncs.com
。Bucket
输入Bucket名称,本教程请填写new-dataworks-workshop。
访问模式
选择Access Key模式。
AccessKey ID
输入访问密钥中的AccessKey ID,本教程请填写LTAI4FvGT3iU4xjKotpUMAjS。
AccessKey Secret
输入访问密钥中的AccessKey Secret,本教程请填写9RSUoRmNxpRC9EhC4m9PjuG7Jzy7px。
重要为方便您更好地体验DataWorks服务,本次模拟实验提供了仅有只读权限的AK供您试用,以上AK仅可用于模拟实验过程中读取样例数据,真实数据生产环境请替换为您自己的数据源AK。
AccessKey Secret只在创建时显示,不支持查看,请妥善保管。如果AccessKey泄露或丢失,请删除并创建新的AccessKey。
单击指定的数据调度资源组连通状态列的测试连通性,等待界面提示测试完成,连通状态为可连通。
重要需确保至少一个资源组为可连通状态,否则此数据源无法使用向导模式创建同步任务。
单击完成。
新建RDS数据源
在数据源管理页面,单击新增数据源。
在新增数据源对话框中,选择数据源类型为MySQL。
在新增MySQL数据源对话框中,配置各项参数。本教程需按示例值填写。
参数
描述
数据源类型
选择连接串模式。
数据源名称
输入数据源名称,本教程请填写rds_workshop_log。
数据源描述
对数据源进行简单描述。
适用环境
如果创建的工作空间为标准模式,勾选开发、生产。开发环境的数据源创建完成后,以同样方式创建生产环境的数据源,否则任务在生产环境执行会报错。
JDBC URL
输入
jdbc:mysql://rm-bp1z69dodhh85z9qa.mysql.rds.aliyuncs.com:3306/workshop
。说明您可以通过公网连接串的形式添加数据源,即通过公网地址访问平台提供的用于教程测试使用的RDS。
用户名
输入用户名,本教程请填写workshop。
密码
输入密码,本教程请填写workshop#2017。
认证选项
无认证。
单击指定的数据调度资源组连通状态列的测试连通性,等待界面提示测试完成,连通状态为可连通。
重要通过实例模式创建的数据源不支持使用公共数据集成资源组跨账号、跨地域等场景同步数据。建议您在实际开发时选择独享数据集成资源组,以获得更多产品能力,保障数据同步的高效性。新增和使用独享数据集成资源组,详情请参见新增和使用独享数据集成资源组。
需确保至少一个资源组为可连通状态,否则此数据源无法使用向导模式创建同步任务。
数据同步时,一个任务只能使用一种资源组。您需要测试每个资源组的连通性,以保证同步任务使用的数据集成资源组能够与数据源连通,否则将无法正常执行数据同步任务。如果您需要同时测试多种资源组,请选中相应资源组后,单击批量测试连通性。详情请参见配置资源组与网络连通。
单击完成。
新建私有OSS数据源
本次实验将EMR引擎的数据存储在OSS数据源中。
在数据源管理页面,单击新增数据源。
在新增数据源对话框中,选择数据源类型为OSS。
在新增OSS数据源对话框中,配置各项参数。
参数
描述
数据源名称
输入数据源的名称,本示例为test。
数据源描述
对数据源进行简单描述。
适用环境
如果创建的工作空间为标准模式,勾选开发、生产。开发环境的数据源创建完成后,以同样方式创建生产环境的数据源,否则任务在生产环境执行会报错。
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。
单击指定资源组连通状态列的测试连通性,等待界面提示测试完成,连通状态为可连通。
重要需确保至少一个资源组为可连通状态,否则此数据源无法使用向导模式创建同步任务。
单击完成。
新建业务流程
单击左上方的图标,选择 。
在数据开发面板,右键单击业务流程,选择新建业务流程。
在新建业务流程对话框中,输入业务名称和描述。
重要业务名称不能超过128个字符,且必须是大小写字母、中文、数字、下划线(_)以及英文句号(.)。
单击新建。
进入业务流程开发面板,鼠标单击虚拟节点并拖拽至右侧的编辑页面。在新建节点对话框中,输入节点名称为workstart,单击确认。
以同样的方式新建两个离线同步节点,节点名称分别为Log2oss和User2oss。
通过拖拽连线,将workstart节点设置为两个离线同步节点的上游节点。
配置workstart节点
在数据开发页面,双击相应业务流程下的虚拟节点。打开该节点的编辑页面,单击右侧的调度配置。
在时间属性区域,设置重跑属性为运行成功或失败皆可重跑。
在调度依赖区域,单击使用工作空间根节点,设置workstart节点的上游节点为工作空间根节点。
由于新版本给每个节点都设置了输入输出节点,所以需要给workstart节点设置一个输入。此处设置其上游节点为工作空间根节点,通常命名为工作空间名_root。
配置完成后,单击左上方的图标。
配置离线同步节点
同步RDS数据源的用户信息至自建的OSS。
在数据开发页面,双击User2oss节点,进入节点配置页面。
网络与资源配置。
完成以下网络与资源配置后,单击下一步,并根据界面提示完成连通性测试。
参数
描述
数据来源
数据来源:MySQL
数据源名称:mysql_workshop_log
我的资源组
选择右下角的
。数据去向
数据去向:OSS
数据源名称:选择前文创建的私有OSS数据源,此处示例为test。
配置数据来源与去向。
参数
描述
数据来源
表:选择数据源中的ods_user_info_d。
切分键:建议使用主键或有索引的列作为切分键,仅支持类型为整型的字段。此处设置切分键为uid。
数据去向
文本类型:选择text类型。
文件名:根据您自建OSS的目录进行输入,示例为ods_user_info_d/user_${bizdate}/user_${bizdate}.txt。其中ods_user_info_d为您自建的目录名。
列分隔符:输入列分隔符为|。
切换脚本模式,补充配置参数。
单击工具栏中的图标,在已有的脚本中:
手动修改writeSingleObject参数取值:"writeSingleObject": true,。
手动添加参数"suffix": ".txt",。
说明writeSingleObject和suffix参数仅支持脚本模式进行添加。
配置完成后,单击工具栏中的图标。
同步OSS数据源的日志信息至自建的OSS。
在数据开发页面,双击Log2oss节点,进入节点配置页面。
网络与资源配置。
完成以下网络与资源配置后,单击下一步,并根据界面提示完成连通性测试。
参数
描述
数据来源
数据来源:OSS
数据源名称:oss_workshop_log
我的资源组
选择右下角的
。数据去向
数据去向:OSS
数据源名称:选择前文创建的私有OSS数据源,此处示例为test。
配置数据来源与去向。
参数
描述
数据来源
文本类型:选择text类型。
文件路径:user_log.txt。
列分隔符:输入列分隔符为|。
压缩格式:包括None、Gzip、Bzip2和Zip四种类型,此处选择None。
数据去向
文本类型:选择text类型。
文件名:根据您自建OSS的目录进行输入,示例为ods_raw_log_d/log_${bizdate}/log_${bizdate}.txt,其中ods_raw_log_d为您自建的目录名。
列分隔符:输入列分隔符为|。
配置字段映射。
重要源数据表中只有一列数据,如果有其他映射过来的空列,您需要手动删除一下。
单击工具栏中的图标,在已有的脚本中手动修改"writeSingleObject": true,并添加"suffix": ".txt",。
说明writeSingleObject和suffix参数仅支持脚本模式进行添加。
配置完成后,单击工具栏中的图标。
新建表
在数据开发页面打开新建的业务流程,右键单击EMR,选择 。
在新建节点对话框中,输入节点名称,单击确认。
此处需要新建两个EMR Hive节点(ods_user_info_d和ods_raw_log_d),分别用于存储同步过来的OSS日志数据和RDS日志数据,并根据下图,通过拉线配置节点的上下游依赖关系。
分别在EMR Hive节点中输入建表语句,并在调度配置中调度资源组选择为独享调度资源组,然后单击保存并运行各建表语句。
新建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为示例路径,需要输入您建立的文件夹的路径名称,其中dw-emr-demo为Bucket名称。
新建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/log_${bizdate}/';
说明上述代码中的location为示例路径,需要输入您建立的文件夹的路径名称。其中dw-emr-demo为Bucket名称。
运行并查看结果
运行业务流程
在数据开发页面,双击业务流程下的WorkShop,打开Workflow业务流程面板后,单击工具栏中的图标,按照上下游依赖关系运行业务流程。
确认执行情况。
查看任务运行状态:节点处于状态,即代表同步执行过程无问题。
查看任务执行日志:右键单击ods_user_info_d、ods_raw_log_d节点,选择查看日志。当日志中出现如下字样,表示同步节点运行成功,并成功同步数据。
查询同步结果
新建临时查询文件
在数据开发页面的左侧导航栏,单击,进入临时查询面板。右键单击临时查询,选择新建节点>EMR Hive。
查询同步结果表。
执行如下SQL语句,确认同步数据写入结果。查看导入ods_raw_log_d和ods_user_info_d的记录数。
--查询语句中的分区列需要更新为业务日期。例如,任务运行的日期为20191107,则业务日期为20191106,即任务运行日期的前一天。 SELECT * from ods_user_info_d where dt=业务日期; SELECT * from ods_raw_log_d where dt=业务日期;
后续步骤
现在,您已经学习了如何进行日志数据同步,完成数据的采集,您可以继续下一个教程。在该教程中,您将学习如何对采集的数据进行计算与分析。详情请参见加工数据。
- 本页导读 (1)