通过DataWorks将Hadoop数据同步到阿里云ES

前提条件

• 已创建Hadoop集群。

  本文使用阿里云E-MapReduce服务自动化搭建Hadoop集群。详细信息，请参见创建集群。

  E-MapReduce Hadoop集群部分配置信息如下：

  • 集群类型：Hadoop

  • 产品版本：EMR-3.26.3

  • 挂载公网：开启

• 已创建阿里云ES实例，并开启实例的自动创建索引功能。具体操作，请参见创建阿里云Elasticsearch实例和配置YML参数。

• 已创建DataWorks工作空间。具体操作，请参见创建工作空间。

费用说明

• ES实例费用，请参见ES计费项。

• 独享数据集成资源组费用，请参见独享数据集成资源组计费说明：包年包月。

步骤一：准备源数据

1. 进入E-MapReduce控制台。

2. 在顶部菜单栏，选择目标地域。

3. 在上方菜单栏，单击数据开发。

4. 在数据开发页面，新建一个数据开发项目，其中资源组选择默认资源组。

   具体操作，请参见项目管理。

5. 在项目列表中，单击目标项目右侧操作列下的作业编辑，新建一个作业。

   具体操作，请参见作业编辑。其中作业类型选择Hive。

6. 创建数据表并插入数据。

   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'

   2. 在运行作业对话框中配置运行参数，单击确定。

      • 资源组：选择默认资源组。

      • 执行集群：选择您已创建的集群。

   3. 重新新建一个作业，输入如下SQL语句，插入测试数据。

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

      insert 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) ;

7. 查看数据是否插入成功。

   1. 新建一个临时查询作业。

      具体操作，请参见临时查询。

   2. 输入如下SQL语句，单击运行。

      select * from hive_esdoc_good_sale where pt =1;

   3. 在页面下方，单击运行记录，再单击操作列下的详情。

   4. 在运维中心，单击作业运行结果。

      此操作可以检查Hadoop集群表中是否已存在数据可用于同步，运行成功后的结果如下。

步骤二：购买并创建独享资源组

购买一个数据集成独享资源组，并为该资源组绑定专有网络和工作空间。独享资源组可以保证数据快速、稳定地传输。

1. 登录DataWorks控制台。

2. 在顶部菜单栏选择相应地域后，在左侧导航栏单击资源组列表。

3. 在独享资源组页签下，单击创建集成资源组。

4. 在DataWorks独享资源购买页面，独享资源类型选择独享数据集成资源，输入资源组名称，单击立即购买，购买独享资源组。

   更多配置信息，请参见购买资源组。

5. 在已创建的独享资源组的操作列，单击网络设置，为该独享资源组绑定专有网络。具体操作，请参见绑定专有网络。

   说明

   本文以独享数据集成资源组通过VPC内网同步数据为例。关于通过公网同步数据，请参见使用独享数据集成资源组执行任务需要在数据库添加的IP白名单。

   独享资源需要与ES实例的专有网络连通才能同步数据。因此需要分别绑定Hadoop集群和ES实例所在的专有网络、可用区和交换机。查看ES实例所在的专有网络、可用区和交换机，请参见查看ES实例的基本信息。

   重要

   绑定专有网络后，您需要将专有网络的交换机网段加入到Hadoop集群、ES实例的VPC私网访问白名单中。具体操作，请参见配置ES实例公网或私网访问白名单。

6. 在页面右上角，单击返回图标，返回资源组列表页面。

7. 在已创建的独享资源组的操作列，单击修改归属工作空间，为该独享资源组绑定目标工作空间。

   具体操作，请参见绑定归属工作空间。

步骤三：添加数据源

1. 进入DataWorks的数据集成页面。

   1. 登录DataWorks控制台。

   2. 在左侧导航栏，单击工作空间列表。

   3. 在目标工作空间的操作列，选择快速进入 数据集成。

2. 在左侧导航栏，单击数据源。

3. 新增Hadoop数据源。

   1. 在数据源列表页面，单击新增数据源。

   2. 在新增数据源页面，搜索并选择HDFS。

   3. 在新增HDFS数据源页面，配置数据源参数。

      配置详情，请参见配置HDFS数据源。

   4. 单击测试连通性，连通状态显示为可连通时，表示连通成功。

   5. 单击完成。

1. 使用同样的方式添加Elasticsearch数据源。配置详情，请参见配置Elasticsearch数据源。

步骤四：配置并运行离线数据同步任务

数据离线同步任务将独享资源组作为一个可以执行任务的资源，独享资源组将获取数据集成服务中数据源的数据，并将数据写入ES。

1. 进入DataWorks的数据开发页面。

   1. 登录DataWorks控制台。

   2. 在左侧导航栏，单击工作空间列表。

   3. 在目标工作空间的操作列，选择快速进入 数据开发。

2. 新建一个离线同步任务。

   1. 在左侧导航栏，选择新建 > 新建业务流程，新建一个业务流程。

   2. 右键单击新建的业务流程，选择新建节点 > 离线同步。

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

3. 配置网络与资源

   1. 在数据来源区域，数据来源选择HDFS，数据源名称选择待同步的数据源名称。

   2. 在我的资源组区域，选择独享资源组。

   3. 在数据去向区域，数据去向选择ES，数据源名称选择待同步的数据源名称。

4. 单击下一步。

5. 配置任务。

   1. 在数据来源区域，选择待同步的表。

   2. 在数据去向区域，配置数据去向的各参数。

   3. 在字段映射区域中，设置来源字段与目标字段的映射关系。具体配置，请参见通过向导模式配置离线同步任务。

   4. 在通道控制区域，配置通道参数。

   详细配置信息，请参见通过向导模式配置离线同步任务。

6. 运行任务。

   1. （可选）配置任务调度属性。在页面右侧，单击调度配置，按照需求配置相应的调度参数。各配置的详细说明请参见调度配置。

   2. 在节点区域的右上角，单击保存图标，保存任务。

   3. 在节点区域的右上角，单击提交图标，提交任务。

      如果您配置了任务调度属性，任务会定期自动执行。您还可以在节点区域的右上角，单击运行图标，立即运行任务。

      运行日志中出现Shell run successfully!表明任务运行成功。

步骤五：验证数据同步结果

1. 登录目标阿里云ES实例的Kibana控制台。

   具体操作，请参见登录Kibana控制台。

2. 在左侧导航栏，单击Dev Tools（开发工具）。

3. 在Console（控制台）中，执行如下命令查看同步的数据。

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

   说明

   hive_esdoc_good_sale为您在数据同步脚本中设置的index字段的值。

   数据同步成功后，返回如下结果。

4. 执行如下命令，搜索品牌为A的所有文档。

   POST /hive_esdoc_good_sale/_search?pretty
   {
     "query": { "match_phrase": { "brand":"品牌A" } }
   }

   

5. 执行如下命令，按照点击次数排序，判断各品牌产品的热度。

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

   

   更多命令和访问方式，请参见阿里云Elasticsearch官方文档和Elasticsearch官方文档。