ES in Hive

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
简介: ElasticSearch已经可以与YARN、Hadoop、Hive、Pig、Spark、Flume等大数据技术框架整合起来使用,尤其是在添加数据的时候,可以使用分布式任务来添加索引数据,尤其是在数据平台上,很多数据存储在Hive中,使用Hive操作ElasticSearch中的数据,将极大的方便开发人员。

ElasticSearch已经可以与YARN、Hadoop、Hive、Pig、Spark、Flume等大数据技术框架整合起来使用,尤其是在添加数据的时候,可以使用分布式任务来添加索引数据,尤其是在数据平台上,很多数据存储在Hive中,使用Hive操作ElasticSearch中的数据,将极大的方便开发人员。这里记录一下Hive与ElasticSearch整合,查询和添加数据的配置使用过程。基于Hive0.13.1、Hadoop-cdh5.0、ElasticSearch 2.1.0。

通过Hive读取与统计分析ElasticSearch中的数据

ElasticSearch中已有的数据

_index:lxw1234
_type:tags
_id:用户ID(cookieid)
字段:area、media_view_tags、interest

elasticsearch

Hive建表

由于我用的ElasticSearch版本为2.1.0,因此必须使用elasticsearch-hadoop-2.2.0才能支持,如果ES版本低于2.1.0,可以使用elasticsearch-hadoop-2.1.2.

下载地址:https://www.elastic.co/downloads/hadoop

 
 
  1. add jar file:///home/liuxiaowen/elasticsearch-hadoop-2.2.0-beta1/dist/elasticsearch-hadoop-hive-2.2.0-beta1.jar;
  2. CREATE EXTERNAL TABLE lxw1234_es_tags (
  3. cookieid string,
  4. area string,
  5. media_view_tags string,
  6. interest string
  7. )
  8. STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
  9. TBLPROPERTIES(
  10. 'es.nodes' = '172.16.212.17:9200,172.16.212.102:9200',
  11. 'es.index.auto.create' = 'false',
  12. 'es.resource' = 'lxw1234/tags',
  13. 'es.read.metadata' = 'true',
  14. 'es.mapping.names' = 'cookieid:_metadata._id, area:area, media_view_tags:media_view_tags, interest:interest');

注意:因为在ES中,lxw1234/tags的_id为cookieid,要想把_id映射到Hive表字段中,必须使用这种方式:
‘es.read.metadata’ = ‘true’,
‘es.mapping.names’ = ‘cookieid:_metadata._id,…’

在Hive中查询数据

elasticsearch

数据已经可以正常查询。

执行SELECT COUNT(1) FROM lxw1234_es_tags;Hive还是通过MapReduce来执行,每个分片使用一个Map任务:

elasticsearch

可以通过在Hive外部表中指定search条件,只查询过滤后的数据。比如,下面的建表语句会从ES中搜索_id=98E5D2DE059F1D563D8565的记录:

 
 
  1. CREATE EXTERNAL TABLE lxw1234_es_tags_2 (
  2. cookieid string,
  3. area string,
  4. media_view_tags string,
  5. interest string
  6. )
  7. STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
  8. TBLPROPERTIES(
  9. 'es.nodes' = '172.16.212.17:9200,172.16.212.102:9200',
  10. 'es.index.auto.create' = 'false',
  11. 'es.resource' = 'lxw1234/tags',
  12. 'es.read.metadata' = 'true',
  13. 'es.mapping.names' = 'cookieid:_metadata._id, area:area, media_view_tags:media_view_tags, interest:interest',
  14. 'es.query' = '?q=_id:98E5D2DE059F1D563D8565'
  15. );
  16.  
  17. hive> select * from lxw1234_es_tags_2;
  18. OK
  19. 98E5D2DE059F1D563D8565 四川|成都 购物|1 购物|1
  20. Time taken: 0.096 seconds, Fetched: 1 row(s)

如果数据量不大,可以使用Hive的Local模式来执行,这样不必提交到Hadoop集群:

在Hive中设置:

 
 
  1. set hive.exec.mode.local.auto.inputbytes.max=134217728;
  2. set hive.exec.mode.local.auto.tasks.max=10;
  3. set hive.exec.mode.local.auto=true;
  4. set fs.defaultFS=file:///;
  5.  
  6. hive> select area,count(1) as cnt from lxw1234_es_tags group by area order by cnt desc limit 20;
  7. Automatically selecting local only mode for query
  8. Total jobs = 2
  9. Launching Job 1 out of 2
  10. …..
  11. Execution log at: /tmp/liuxiaowen/liuxiaowen_20151211133030_97b50138-d55d-4a39-bc8e-cbdf09e33ee6.log
  12. Job running in-process (local Hadoop)
  13. Hadoop job information for null: number of mappers: 0; number of reducers: 0
  14. 2015-12-11 13:30:59,648 null map = 100%, reduce = 100%
  15. Ended Job = job_local1283765460_0001
  16. Execution completed successfully
  17. MapredLocal task succeeded
  18. OK
  19. 北京|北京 10
  20. 四川|成都 4
  21. 重庆|重庆 3
  22. 山西|太原 3
  23. 上海|上海 3
  24. 广东|深圳 3
  25. 湖北|武汉 2
  26. 陕西|西安 2
  27. 福建|厦门 2
  28. 广东|中山 2
  29. 福建|三明 2
  30. 山东|济宁 2
  31. 甘肃|兰州 2
  32. 安徽|合肥 2
  33. 湖南|长沙 2
  34. 湖南|湘西 2
  35. 河南|洛阳 2
  36. 江苏|南京 2
  37. 黑龙江|哈尔滨 2
  38. 广西|南宁 2
  39. Time taken: 13.037 seconds, Fetched: 20 row(s)
  40. hive>

很快完成了查询与统计。

通过Hive向ElasticSearch中写数据

Hive建表

 
 
  1. add jar file:///home/liuxiaowen/elasticsearch-hadoop-2.2.0-beta1/dist/elasticsearch-hadoop-hive-2.2.0-beta1.jar;
  2. CREATE EXTERNAL TABLE lxw1234_es_user_tags (
  3. cookieid string,
  4. area string,
  5. gendercode STRING,
  6. birthday STRING,
  7. jobtitle STRING,
  8. familystatuscode STRING,
  9. haschildrencode STRING,
  10. media_view_tags string,
  11. order_click_tags STRING,
  12. search_egine_tags STRING,
  13. interest string )
  14. STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
  15. TBLPROPERTIES(
  16. 'es.nodes' = '172.16.212.17:9200,172.16.212.102:9200',
  17. 'es.index.auto.create' = 'true',
  18. 'es.resource' = 'lxw1234/user_tags',
  19. 'es.mapping.id' = 'cookieid',
  20. 'es.mapping.names' = 'area:area,
  21. gendercode:gendercode,
  22. birthday:birthday,
  23. jobtitle:jobtitle,
  24. familystatuscode:familystatuscode,
  25. haschildrencode:haschildrencode,
  26. media_view_tags:media_view_tags,
  27. order_click_tags:order_click_tags,
  28. search_egine_tags:search_egine_tags,
  29. interest:interest');

这里要注意下:如果是往_id中插入数据,需要设置’es.mapping.id’ = ‘cookieid’参数,表示Hive中的cookieid字段对应到ES中的_id,而es.mapping.names中不需要再映射,这点和读取时候的配置不一样。

关闭Hive推测执行,执行INSERT:

 
 
  1. SET hive.mapred.reduce.tasks.speculative.execution = false;
  2. SET mapreduce.map.speculative = false;
  3. SET mapreduce.reduce.speculative = false;
  4.  
  5. INSERT overwrite TABLE lxw1234_es_user_tags
  6. SELECT cookieid,
  7. area,
  8. gendercode,
  9. birthday,
  10. jobtitle,
  11. familystatuscode,
  12. haschildrencode,
  13. media_view_tags,
  14. order_click_tags,
  15. search_egine_tags,
  16. interest
  17. FROM source_table;

注意:如果ES集群规模小,而source_table数据量特别大、Map任务数太多的时候,会引发错误:

 
 
  1. Caused by: org.elasticsearch.hadoop.rest.EsHadoopInvalidRequest:
  2. FOUND unrecoverable error [172.16.212.17:9200] returned Too Many Requests(429) - rejected
  3. execution of org.elasticsearch.action.support.replication.TransportReplicationAction$PrimaryPhase$1@b6fa90f
  4. ON EsThreadPoolExecutor[bulk, queue capacity = 50,
  5. org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@22e73289[Running, pool size = 32, active threads = 32, queued tasks = 52, completed tasks = 12505]];
  6. Bailing out..

原因是Map任务数太多,并发发送至ES的请求数过多。
这个和ES集群规模以及bulk参数设置有关,目前还没弄明白。
减少source_table数据量(即减少Map任务数)之后,没有出现这个错误。

执行完成后,在ES中查询lxw1234/user_tags的数据:

 
 
  1. curl -XGET http://172.16.212.17:9200/lxw1234/user_tags/_search?pretty -d '
  2. {
  3. "query" : {
  4. "match" : {
  5. "area" : "成都"
  6. }
  7. }
  8. }'

elasticsearch

数据已经写入到ElasticSearch中。

总结

使用Hive将数据添加到ElasticSearch中还是非常实用的,因为我们的数据都是在HDFS上,通过Hive可以查询的。

另外,通过Hive可以查询ES数据,并在其上做复杂的统计与分析,但性能一般,比不上使用ES原生API,亦或是还没有掌握使用技巧,后面继续研究。

相关阅读:

ElasticSearch集群安装配置

ElasticSearch与Hive整合官方文档

相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
目录
相关文章
|
6月前
|
SQL 数据采集 数据挖掘
大数据行业应用之Hive数据分析航班线路相关的各项指标
大数据行业应用之Hive数据分析航班线路相关的各项指标
191 1
|
SQL Java 大数据
Hive实战(03)-深入了解Hive JDBC:在大数据世界中实现数据交互
Hive实战(03)-深入了解Hive JDBC:在大数据世界中实现数据交互
607 1
|
SQL 分布式计算 大数据
黑马程序员-大数据入门到实战-分布式SQL计算 Hive 入门
黑马程序员-大数据入门到实战-分布式SQL计算 Hive 入门
154 0
|
6月前
|
SQL 存储 大数据
【大数据技术Hadoop+Spark】Hive基础SQL语法DDL、DML、DQL讲解及演示(附SQL语句)
【大数据技术Hadoop+Spark】Hive基础SQL语法DDL、DML、DQL讲解及演示(附SQL语句)
253 0
|
SQL 存储 大数据
黑马程序员-大数据入门到实战-分布式SQL计算 Hive 语法与概念
黑马程序员-大数据入门到实战-分布式SQL计算 Hive 语法与概念
142 0
|
6月前
|
SQL 分布式计算 数据库
【大数据技术Spark】Spark SQL操作Dataframe、读写MySQL、Hive数据库实战(附源码)
【大数据技术Spark】Spark SQL操作Dataframe、读写MySQL、Hive数据库实战(附源码)
250 0
|
6月前
|
SQL 存储 分布式计算
【大数据技术Hadoop+Spark】Hive数据仓库架构、优缺点、数据模型介绍(图文解释 超详细)
【大数据技术Hadoop+Spark】Hive数据仓库架构、优缺点、数据模型介绍(图文解释 超详细)
1071 0
|
1月前
|
SQL 分布式计算 Java
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
大数据-96 Spark 集群 SparkSQL Scala编写SQL操作SparkSQL的数据源:JSON、CSV、JDBC、Hive
35 0
|
4月前
|
SQL 分布式计算 大数据
大数据处理平台Hive详解
【7月更文挑战第15天】Hive作为基于Hadoop的数据仓库工具,在大数据处理和分析领域发挥着重要作用。通过提供类SQL的查询语言,Hive降低了数据处理的门槛,使得具有SQL背景的开发者可以轻松地处理大规模数据。然而,Hive也存在查询延迟高、表达能力有限等缺点,需要在实际应用中根据具体场景和需求进行选择和优化。