轉(zhuǎn)自：http://lxw1234.com/archives/2015/12/585.htm
關(guān)鍵字：hive、elasticsearch、integration、整合
ElasticSearch已經(jīng)可以與YARN、Hadoop、Hive、Pig、Spark、Flume等大數(shù)據(jù)技術(shù)框架整合起來使用，尤其是在添加數(shù)據(jù)的時(shí)候，可以使用分布式任務(wù)來添加索引數(shù)據(jù)，尤其是在數(shù)據(jù)平臺(tái)上，很多數(shù)據(jù)存儲(chǔ)在Hive中，使用Hive操作ElasticSearch中的數(shù)據(jù)，將極大的方便開發(fā)人員。這里記錄一下Hive與ElasticSearch整合，查詢和添加數(shù)據(jù)的配置使用過程?；贖ive0.13.1、Hadoop-cdh5.0、ElasticSearch 2.1.0。
通過Hive讀取與統(tǒng)計(jì)分析ElasticSearch中的數(shù)據(jù)
ElasticSearch中已有的數(shù)據(jù)
_index：lxw1234
_type：tags
_id：用戶ID（cookieid）
字段：area、media_view_tags、interest
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
add jar file:///home/liuxiaowen/elasticsearch-hadoop-2.2.0-beta1/dist/elasticsearch-hadoop-hive-2.2.0-beta1.jar;
CREATE EXTERNAL TABLE lxw1234_es_tags(
cookieidstring,
areastring,
media_view_tagsstring,
intereststring
)
STORED BY'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES(
'es.nodes'='172.16.212.17:9200,172.16.212.102:9200',
'es.index.auto.create'='false',
'es.resource'='lxw1234/tags',
'es.read.metadata'='true',
'es.mapping.names'='cookieid:_metadata._id, area:area, media_view_tags:media_view_tags, interest:interest');
注意：因?yàn)樵贓S中，lxw1234/tags的_id為cookieid，要想把_id映射到Hive表字段中，必須使用這種方式：
‘es.read.metadata’ = ‘true’,
‘es.mapping.names’ = ‘cookieid:_metadata._id,…’
在Hive中查詢數(shù)據(jù)
數(shù)據(jù)已經(jīng)可以正常查詢。
執(zhí)行SELECT COUNT(1) FROM lxw1234_es_tags;Hive還是通過MapReduce來執(zhí)行，每個(gè)分片使用一個(gè)Map任務(wù)：
可以通過在Hive外部表中指定search條件，只查詢過濾后的數(shù)據(jù)。比如，下面的建表語句會(huì)從ES中搜索_id=98E5D2DE059F1D563D8565的記錄：
CREATE EXTERNAL TABLE lxw1234_es_tags_2(
cookieidstring,
areastring,
media_view_tagsstring,
intereststring
)
STORED BY'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES(
'es.nodes'='172.16.212.17:9200,172.16.212.102:9200',
'es.index.auto.create'='false',
'es.resource'='lxw1234/tags',
'es.read.metadata'='true',
'es.mapping.names'='cookieid:_metadata._id, area:area, media_view_tags:media_view_tags, interest:interest',
'es.query'='?q=_id:98E5D2DE059F1D563D8565'
);
hive>select*fromlxw1234_es_tags_2;
OK
98E5D2DE059F1D563D8565四川|成都購物|1購物|1
Timetaken:0.096seconds,Fetched:1row(s)
如果數(shù)據(jù)量不大，可以使用Hive的Local模式來執(zhí)行，這樣不必提交到Hadoop集群：
在Hive中設(shè)置：
sethive.exec.mode.local.auto.inputbytes.max=134217728;
sethive.exec.mode.local.auto.tasks.max=10;
sethive.exec.mode.local.auto=true;
setfs.defaultFS=file:///;
hive>selectarea,count(1)ascntfromlxw1234_es_tagsgroupbyarea orderbycnt desc limit20;
Automaticallyselectinglocalonly modeforquery
Totaljobs=2
LaunchingJob1outof2
…..
Executionlog at:/tmp/liuxiaowen/liuxiaowen_20151211133030_97b50138-d55d-4a39-bc8e-cbdf09e33ee6.log
Jobrunningin-process(localHadoop)
Hadoopjob informationfornull:number of mappers:0;number of reducers:0
2015-12-1113:30:59,648nullmap=100%,reduce=100%
EndedJob=job_local1283765460_0001
Executioncompleted successfully
MapredLocaltask succeeded
OK
北京|北京10
四川|成都4
重慶|重慶3
山西|太原3
上海|上海3
廣東|深圳3
湖北|武漢2
陜西|西安2
福建|廈門2
廣東|中山2
福建|三明2
山東|濟(jì)寧2
甘肅|蘭州2
安徽|合肥2
湖南|長沙2
湖南|湘西2
河南|洛陽2
江蘇|南京2
黑龍江|哈爾濱2
廣西|南寧2
Timetaken:13.037seconds,Fetched:20row(s)
hive>
很快完成了查詢與統(tǒng)計(jì)。
通過Hive向ElasticSearch中寫數(shù)據(jù)
Hive建表
add jar file:///home/liuxiaowen/elasticsearch-hadoop-2.2.0-beta1/dist/elasticsearch-hadoop-hive-2.2.0-beta1.jar;
CREATE EXTERNAL TABLE lxw1234_es_user_tags(
cookieidstring,
areastring,
gendercode STRING,
birthday STRING,
jobtitle STRING,
familystatuscode STRING,
haschildrencode STRING,
media_view_tagsstring,
order_click_tags STRING,
search_egine_tags STRING,
intereststring)
STORED BY'org.elasticsearch.hadoop.hive.EsStorageHandler'
TBLPROPERTIES(
'es.nodes'='172.16.212.17:9200,172.16.212.102:9200',
'es.index.auto.create'='true',
'es.resource'='lxw1234/user_tags',
'es.mapping.id'='cookieid',
'es.mapping.names'='area:area,
gendercode:gendercode,
birthday:birthday,
jobtitle:jobtitle,
familystatuscode:familystatuscode,
haschildrencode:haschildrencode,
media_view_tags:media_view_tags,
order_click_tags:order_click_tags,
search_egine_tags:search_egine_tags,
interest:interest');
這里要注意下：如果是往_id中插入數(shù)據(jù)，需要設(shè)置’es.mapping.id’ = ‘cookieid’參數(shù)，表示Hive中的cookieid字段對(duì)應(yīng)到ES中的_id，而es.mapping.names中不需要再映射，這點(diǎn)和讀取時(shí)候的配置不一樣。
關(guān)閉Hive推測執(zhí)行,執(zhí)行INSERT：
SET hive.mapred.reduce.tasks.speculative.execution=false;
SET mapreduce.map.speculative=false;
SET mapreduce.reduce.speculative=false;
INSERT overwrite TABLE lxw1234_es_user_tags
SELECT cookieid,
area,
gendercode,
birthday,
jobtitle,
familystatuscode,
haschildrencode,
media_view_tags,
order_click_tags,
search_egine_tags,
interest
FROM source_table;
注意：如果ES集群規(guī)模小，而source_table數(shù)據(jù)量特別大、Map任務(wù)數(shù)太多的時(shí)候，會(huì)引發(fā)錯(cuò)誤：
Causedby:org.elasticsearch.hadoop.rest.EsHadoopInvalidRequest:
FOUND unrecoverable error[172.16.212.17:9200]returnedTooManyRequests(429)-rejected
execution of org.elasticsearch.action.support.replication.TransportReplicationAction$PrimaryPhase$1@b6fa90f
ONEsThreadPoolExecutor[bulk,queue capacity=50,
org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@22e73289[Running,pool size=32,active threads=32,queued tasks=52,completed tasks=12505]];
Bailingout..
原因是Map任務(wù)數(shù)太多，并發(fā)發(fā)送至ES的請(qǐng)求數(shù)過多。
這個(gè)和ES集群規(guī)模以及bulk參數(shù)設(shè)置有關(guān)，目前還沒弄明白。
減少source_table數(shù)據(jù)量（即減少M(fèi)ap任務(wù)數(shù)）之后，沒有出現(xiàn)這個(gè)錯(cuò)誤。
執(zhí)行完成后，在ES中查詢lxw1234/user_tags的數(shù)據(jù)：
curl-XGET http://172.16.212.17:9200/lxw1234/user_tags/_search?pretty -d '
{
"query":{
"match":{
"area":"成都"
}
}
}'
數(shù)據(jù)已經(jīng)寫入到ElasticSearch中。
總結(jié)
使用Hive將數(shù)據(jù)添加到ElasticSearch中還是非常實(shí)用的，因?yàn)槲覀兊臄?shù)據(jù)都是在HDFS上，通過Hive可以查詢的。
另外，通過Hive可以查詢ES數(shù)據(jù)，并在其上做復(fù)雜的統(tǒng)計(jì)與分析，但性能一般，比不上使用ES原生API，亦或是還沒有掌握使用技巧，后面繼續(xù)研究。
相關(guān)閱讀：
