【最佳实践】如何运用DataWorks数据同步功能,将Hadoop数据同步到阿里云Elasticsearch上

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
简介: 如何通过DataWorks数据同步功能,将Hadoop数据同步到阿里云Elasticsearch上,并进行搜索分析。

简介:

本文向您详细介绍如何通过DataWorks数据同步功能,将Hadoop数据同步到阿里云Elasticsearch上,并进行搜索分析。

本文字数:2673
阅读时间:预计10分钟

目录

背景信息

环境准备
数据准备
数据同步
结果验证
数据搜索与分析

以下是正文


背景信息

您也可以使用Java代码进行同步,具体请参考通过ES-Hadoop将Hadoop数据写入阿里云Elasticsearch和在E-MapReduce中使用ES-Hadoop

环境准备

  1. 搭建Hadoop集群。在进行数据同步前,您需要保证自己的Hadoop集群环境正常。本文使用阿里云EMR服务自动化搭建Hadoop集群,详细过程请参见步骤三:创建集群。EMR Hadoop的版本信息如下。
  • EMR版本:EMR-3.11.0
  • 集群类型:HADOOP
  • 软件信息:HAFS2.7.2/YARN2.7.2/Hive2.3.3/Ganglia3.7.2/Spark2.3.1/HUE4.1.0/Zeppelin0.8.0/ Tez0.9.1 / Sqoop1.4.7 / Pig0.14.0 / ApacheDS2.0.0 / Knox0.13.0
  1. Hadoop集群使用VPC网络,区域为华东1(杭州),主实例组ECS计算ziyu 配置公网及内网IP,高可用选择为否(非HA模式),具体配置如下图所示

image.png

  1. 购买和配置Elasticsearch。登录Elasticsearch控制台,参考购买和配置,购买一个Elasticsearch实例。选择与EMR集群相同的区域和VPC网络配置,如下图所示。

    image.png

  2. 创建DataWorks工作空间。创建DataWorks项目,区域选择华东1区。本文直接使用已经存在的项目bigdata_DOC。
    image.png

数据准备

在Hadoop集群中创建测试数据,步骤如下。

  1. 进入EMR控制台界面,单击左侧菜单栏的交互式工作台
  2. 选择文件 > 新建交互式任务
  3. Notebook对话框中,输入交互式任务的名称,选择默认类型以及关联集群,单击确认。本文新建一个名为es_test_hive的交互式任务,默认类型Hive关联集群环境准备中创建的EMR Hadoop集群。

image.png

  1. 在代码编辑区域中,输入Hive建表语句,单击运行。本文档使用的建表语句如下。
CREATE TABLE IF NOT

EXISTS hive_esdoc_good_sale(
 create_time timestamp,
 category STRING,
 brand STRING,
 buyer_id STRING,
 trans_num BIGINT,
 trans_amount DOUBLE,
 click_cnt BIGINT
 )
 PARTITIONED BY (pt string) ROW FORMAT
DELIMITED FIELDS TERMINATED BY ',' lines terminated by '\n'

表创建成功后,系统会提示Query executed successfully。

image.png

  1. 单击文件 > 新建段落,在段落编辑区域输入SQL语句,单击运行,插入测试数据。
    image.png

您可以选择从OSS或其他数据源导入测试数据,也可以手动插入少量的测试数据。本文使用手动插入数据的方法,脚本如下。

jinsert into
hive_esdoc_good_sale PARTITION(pt =1 ) values('2018-08-21','外套','品牌A','lilei',3,500.6,7),('2018-08-22','生鲜','品牌B','lilei',1,303,8),('2018-08-22','外套','品牌C','hanmeimei',2,510,2),(2018-08-22,'卫浴','品牌A','hanmeimei',1,442.5,1),('2018-08-22','生鲜','品牌D','hanmeimei',2,234,3),('2018-08-23','外套','品牌B','jimmy',9,2000,7),('2018-08-23','生鲜','品牌A','jimmy',5,45.1,5),('2018-08-23','外套','品牌E','jimmy',5,100.2,4),('2018-08-24','生鲜','品牌G','peiqi',10,5560,7),('2018-08-24','卫浴','品牌F','peiqi',1,445.6,2),('2018-08-24','外套','品牌A','ray',3,777,3),('2018-08-24','卫浴','品牌G','ray',3,122,3),('2018-08-24','外套','品牌C','ray',1,62,7) ;
  1. 使用同样的方式新建段落,并在段落编辑区域输入`js
    • from hive_esdoc_good_sale where pt =1;
语句,单击**运行**。
此操作可以检查Hadoop集群表中是否已存在数据可用于同步,运行成功结果如下。

![image.png](https://ucc.alicdn.com/pic/developer-ecology/8ea1e62da59d42da850f0f8ab2452578.png)

### 数据同步


> **说明
**
由于DataWorks项目所处的网络环境与Hadoop集群中的数据节点(Data Node)网络通常不可达,因此您可以通过自定义资源组的方式,将DataWorks的同步任务运行在Hadoop集群的Master节点上(Hadoop集群内Master节点和数据节点通常可达)。


1. 查看Hadoop集群的数据节点。
- 在EMR控制台上,单击左侧菜单栏的集群。
-  选择您的集群,单击右侧的管理。
- 在集群管理控制台上,单击左侧菜单栏的主机列表,查看集群master节点和数据节点信息。

![image.png](https://ucc.alicdn.com/pic/developer-ecology/cfbdeccef47a4511bb2132b39fd44ce2.png)

> **说明
**
通常非HA模式的EMR上Hadoop集群的Master节点主机名为emr-header-1,Data Node主机名为emr-worker-X。

- 单击上图中Master节点的ECS ID,进入ECS实例详情页。单击远程连接进入ECS服务器,通过hadoop dfsadmin -report命令查看数据节点信息。

![image.png](https://ucc.alicdn.com/pic/developer-ecology/b9e6ba7733d0483bbb082bc09bf3061d.png)


2. 新建自定义资源组。

- 进入DataWorks的**数据集成**页面,选择**资源组** > **新增资源组**。
![image.png](https://ucc.alicdn.com/pic/developer-ecology/7dbd54d52e3b40d1b5d3d3426db7b4ba.png)
关于自定义资源组的详细信息请参见[新增任务资源](https://help.aliyun.com/document_detail/72979.html?spm=a2c4g.11186623.2.31.341d491ecP7i8I#concept-wfz-j45-q2b)。

- 根据界面提示,输入资源组名称和服务器信息。此服务器为您EMR集群的Master节点,服务器信息说明如下。
![image.png](https://ucc.alicdn.com/pic/developer-ecology/cd02fb0498eb4207b76babf76baacfba.png)
​
 ![image.png](https://ucc.alicdn.com/pic/developer-ecology/17ccc71e656d405aa3f5d31cbcf00b94.png)




> **注意 **完成添加服务器后,您需要保证Master Node与DataWorks网络可达。
- 如果您使用的是ECS服务器,需设置服务器的安全组。
- 如果您使用的内网IP互通,需要[添加安全组。](https://help.aliyun.com/document_detail/72978.html?spm=a2c4g.11186623.2.33.341d491ecP7i8I#concept-ec4-cj5-q2b)
- 如果您使用的是公网IP,可直接设置安全组公网出入方向规则。
由于本文档的EMR集群使用的是VPC网络,且与DataWorks在同一区域下,因此不需要进行安全组设置。  

- 按照提示安装自定义资源组Agent。

> **注意**:由于本文使用的是VPC网络类型,因此不需开通8000端口。  

观察到当前状态为**可用**时,说明新增自定义资源组成功。如果状态为不可用,您可以登录Master Node,使用```js
tail –f/home/admin/alisatasknode/logs/heartbeat.log
命```  
令查看DataWorks与Master Node之间心跳报文是否超时。

![image.png](https://ucc.alicdn.com/pic/developer-ecology/e9537ee6ae4048a0828867a6d5c232a4.png)

3. 新建数据源
- 在DataWorks的数据集成页面,单击**数据源**>**新增数据源**,在弹框中选择**HDFS**类型的数据源
![image.png](https://ucc.alicdn.com/pic/developer-ecology/ed40f8ee2f9049d4929e81b5aa007784.png)

- 在**新增HDFS**数据源页面中,填写**数据源名称**和**defaultFS。**
![image.png](https://ucc.alicdn.com/pic/developer-ecology/db89e0e4832347f791158a3a03aecced.png)


> 注意 对于EMR Hadoop集群而言,如果Hadoop集群为非HA集群,则此处地址为```js
hdfs://emr-header-1的IP:9000
。```  
如果Hadoop集群为HA集群,则此处地址为```js
hdfs://emr-header-1的IP:8020
。```  
在本文中,emr-header-1与DataWorks通过VPC网络连接,因此此处填写内网IP,且不支持连通性测试。  

4. 配置数据同步任务。
a. 在DataWorks数据集成页面,单击左侧菜单栏的**同步任务**,选择**新建** > **脚本模式**。
b. 在**导入模板**对话框中,选择数据源类型如下,单击**确认**。
![image.png](https://ucc.alicdn.com/pic/developer-ecology/f8c22f4f6d484e62b29fd1d5b6359b7e.png)
c. 完成导入模板后,同步任务会转入[脚本模式](https://help.aliyun.com/document_detail/74304.html?spm=a2c4g.11186623.2.38.341d491ecP7i8I#concept-olb-drc-p2b),本文中配置脚本如下,相关解释请参见脚本模式配置,Elasticsearch的配置规则请参考[配置Elasticsearch Writer。](https://help.aliyun.com/document_detail/74362.html?spm=a2c4g.11186623.2.39.341d491ecP7i8I#concept-okj-c24-q2b)
![image.png](https://ucc.alicdn.com/pic/developer-ecology/dea407d18bd84b948a4e4e7efbac333e.png)
- 同步脚本的配置分为三个部分,Reader用来配置您上游数据源(待同步数据的云产品)的config,Writer用来配置 Elasticsearch的config,setting用来配置同步中的一些丢包和最大并发等。
- path为数据在Hadoop集群中存放的位置,您可以在登录master node后,```js
使用hdfs dfs –ls /user/hive/warehouse/hive_esdoc_good_sale
命```  
令确认。对于分区表,您可以不指定分区,DataWorks数据同步会自动递归到分区路径。
- 由于Elasticsearch不支持timestamp类型,本文档将**creat_time**字段的类型设置为string。
- **endpoint**为Elasticsearch 的内网或外网地址。如果您使用的是内网地址,请在Elasticsearch的集群配置页面,配置Elasticsearch的系统白名单。如果您是用的是外网地址,请在Elasticsearch的网络配置页面,配置 Elasticsearch的公网地址访问白名单(包括DataWorks服务器的IP地址和您所使用的资源组的IP地址)。
- Elasticsearch Writer中**accessId**和**accessKey**需要配置您的Elasticsearch的访问用户名(默认为elastic)和密码。
- **index**为Elasticsearch实例的索引,您需要使用该索引名称访问Elasticsearch的数据。
- 在创建同步任务时,DataWorks的默认配置脚本中,**errorLimit**的**record**字段值为0,您需要将其修改为大一些的数值,比如1000。

d. 完成配置后,单击页面右侧的**配置任务资源组**,选择您创建的资源组名称,完成后单击**运行**。如果提示**任务运行成功**,则说明同步任务已完成。如果运行失败,可通过复制日志进行进一步排查。

### 结果验证

1. 进入[Elasticsearch控制台](https://elasticsearch.console.aliyun.com/?spm=a2c4g.11186623.2.41.341d491ecP7i8I),单击实例名称>可视化控制,在Kibana区域中,单击右下角进入控制台。
2. 输入用户名和密码,单击登录进入kibana控制台,选择Dev Tools。
3. 在**Console**控制台中,执行如下命令,查看已经同步过来的数据

POST /hive_doc_esgood_sale/_search?pretty
{
"query": { "match_all": {}}
}



```hive_doc_esgood_sale```  为您同步数据时,设置的index字段的值。


![image.png](https://ucc.alicdn.com/pic/developer-ecology/be8ec07abb164909a9c9ff89417e7bfb.png)



### 数据搜索与分析
1. 在**Console**控制台中,执行如下命令,返回品牌为A的所有文档。

POST /hive_doc_esgood_sale/_search?pretty
{
"query": { "match_all": {} },
"sort": { "click_cnt": { "order": "desc" } },
"_source": ["category", "brand","click_cnt"]
}

![image.png](https://ucc.alicdn.com/pic/developer-ecology/2de8fea7eeb34011a26e7f729b98baeb.png)
2. 在**Console**控制台中,执行如下命令,按照**点击次数**进行排序,判断各品牌产品的热度。

POST /hive_doc_esgood_sale/_search?pretty
{
"query": { "match_all": {} },
"sort": { "click_cnt": { "order": "desc" } },
"_source": ["category", "brand","click_cnt"]
}
`
image.png

更多命令和访问方式,请参见阿里云Elasticsearch官方文档Elastic.co官方帮助中心

加入我们

加入《Elasticsearch中文技术社区》,与更多开发者探讨交流

8.jpeg


订阅《阿里云Elasticsearch技术交流期刊》每月定期为大家推送相关干货

9.png


2019年阿里云云栖大会上,Elasticsearch背后的商业公司Elastic与阿里云Elasticsearch确定战略合作升级,在100%兼容开源的基础上,完成了ELK的完整生态云上闭环,欢迎开通使用。
点击了解更多产品信息
相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
一站式大数据开发治理平台DataWorks初级课程
DataWorks 从 2009 年开始,十ー年里一直支持阿里巴巴集团内部数据中台的建设,2019 年双 11 稳定支撑每日千万级的任务调度。每天阿里巴巴内部有数万名数据和算法工程师正在使用DataWorks,承了阿里巴巴 99%的据业务构建。本课程主要介绍了阿里巴巴大数据技术发展历程与 DataWorks 几大模块的基本能力。 课程目标  通过讲师的详细讲解与实际演示,学员可以一边学习一边进行实际操作,可以深入了解DataWorks各大模块的使用方式和具体功能,让学员对DataWorks数据集成、开发、分析、运维、安全、治理等方面有深刻的了解,加深对阿里云大数据产品体系的理解与认识。 适合人群  企业数据仓库开发人员  大数据平台开发人员  数据分析师  大数据运维人员  对于大数据平台、数据中台产品感兴趣的开发者
相关文章
|
7天前
|
弹性计算 运维 监控
解密阿里云弹性计算:探索云服务器ECS的核心功能
阿里云ECS是核心计算服务,提供弹性云服务器资源,支持实例按需配置、集群管理和监控,集成安全防护,确保服务稳定、安全,助力高效业务运营。
227 0
|
7天前
|
存储 监控 API
Elasticsearch 8.X Rollup 功能详解及避坑指南
Elasticsearch 8.X Rollup 功能详解及避坑指南
22 0
|
5天前
|
存储 SQL 分布式计算
|
7天前
|
分布式计算 DataWorks Oracle
DataWorks操作报错合集之DataWorks ODPS数据同步后,timesramp遇到时区问题,解决方法是什么
DataWorks是阿里云提供的一站式大数据开发与治理平台,支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
28 0
|
7天前
|
缓存 DataWorks 监控
DataWorks操作报错合集之在DataWorks中进行数据同步时,遇到数据量大的表同步时报链接异常,该怎么办
DataWorks是阿里云提供的一站式大数据开发与治理平台,支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中,可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
22 0
|
7天前
|
SQL JSON 分布式计算
DataWorks产品使用合集之DataWorks一键maxcompute数据同步的操作步骤是什么
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
27 0
|
7天前
|
存储 分布式计算 DataWorks
DataWorks产品使用合集之在DataWorks中,从数据同步到最终的大屏幕展示的基本流程如何解决
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
29 0
DataWorks产品使用合集之在DataWorks中,从数据同步到最终的大屏幕展示的基本流程如何解决
|
7天前
|
DataWorks Shell 对象存储
DataWorks产品使用合集之在 DataWorks 中,有一个 MySQL 数据表,数据量非常大且数据会不断更新将这些数据同步到 DataWorks如何解决
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
33 3
|
7天前
|
数据采集 分布式计算 DataWorks
DataWorks产品使用合集之在阿里云的DataWorks中,使用CREATE AS创建分区表的步骤如何解决
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
32 2
|
7天前
|
分布式计算 DataWorks 调度
DataWorks产品使用合集之阿里云的DataWorks中,提交手动业务流程的步骤如何解决
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
25 0

相关产品

  • 检索分析服务 Elasticsearch版