Hive整合ES实现ES与Hdfs数据比对

• Hive整合ES实现ES与Hdfs数据比对
• • 环境介绍
  • 准备Yarn和客户端环境
  • 在Hive中建表&插入数据
  • • 创建ES索引
    • 基于ES索引创建Hive外部表
    • • 创建表
      • 插入数据
    • 创建基于Hdfs外部表
    • • 创建表
      • 插入数据
  • 数据比对
  • 总结
• 客官且慢，点赞、收藏+关注 谢谢~

业务场景中需要ES数据和Hdfs数据进行比对，实现数据查漏补缺。为此需要将Hdfs与ES进行整合，从而引入elasticsearch-hadoop，目前elasticsearch-hadoop支持Map/Reduce、Hive、Pig、Spark和Storm，以及整合Hdfs用于存储ES备份数据。

因此可以在Hive中建立一张ES外部表和一张Hdfs外部表，实现将Hive数据导入ES、Hive数据与ES数据比对、ES数据下载到Hive等等。

Java版本：

java -version
java version "1.8.0_181"

查看Hadoop版本:

hadoop version
Hadoop 3.0.0-cdh6.2.0

查看Hive版本：

hive --version
Hive 2.1.1-cdh6.2.0

ES版本:

6.2.3

Elasticsearch-Hadoop包版本：(由于ES集群环境是6.2.3，只能用这个版本试一下，不过很幸运，依然奏效)

elasticsearch-hadoop-7.15.2

获得elasticsearch-hadoop-7.15.2.jar包：

下载地址

设置任务提交相关依赖：

hive -hiveconf hive.aux.jars.path=/root/elasticsearch-hadoop-7.15.2.jar

设置Hive客户端依赖：

ADD JAR /root/elasticsearch-hadoop-7.15.2.jar;

PUT my_index_0001
{
  "mappings": {
    "doc": {
      "properties": {
        "xb_mc": {
          "type": "text",
          "analyzer": "keyword"
        },
        "hjdqh_dm": {
          "type": "text",
          "analyzer": "keyword"
        },
        "zjhm": {
          "type": "text",
          "analyzer": "keyword"
        }
      }
    }
  }
}

# 先进入hive shell client
hive -hiveconf hive.aux.jars.path=/root/elasticsearch-hadoop-7.15.2.jar
ADD JAR /root/elasticsearch-hadoop-7.15.2.jar;

CREATE EXTERNAL TABLE es_my_index_0001 (
    xb_mc      STRING,    zjhm      STRING,
    hjdqh_dm    STRING)
    STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler' 
    TBLPROPERTIES(
        'es.resource' = 'es_my_index_0001/doc',
        'es.query' = '{  "query": {"match_all": {}}}',
        'es.nodes' = 'ES_cluster_ip','es.mapping.id'='zjhm'
    );  

insert into es_my_index_0001 values("01","428008199901010898","test-1");
insert into es_my_index_0001 values("01","428008199901010897","test-1");

注意

ES_cluster_ip需要填写自己ES集群的ip，填写一个就行，默认会进行自动发现其他节点；'es.mapping.id'='zjhm'表示用字段zjhm作为ES文档的文档id；更多其他配置信息可以参考博客和官网

CREATE EXTERNAL TABLE test_table
(   xb_mc      STRING,    zjhm      STRING,
    hjdqh_dm    STRING
) STORED AS PARQUET LOCATION 'hdfs://ha-nameservice/hive/test_table';

insert into test_table values("1","428008199901010898","test");
insert into test_table values("01","428008199901010897","test-1");
insert into test_table values("01","428008199901010871","test-1");

在hive中运行如下sql，即可获得ES中漏写入数据

select a.zjhm ,b.zjhm from test_table a left join (select zjhm from es_my_index_0001) b on a.zjhm= b.zjhm  where b.zjhm is null;

运行结果：

OK
428008199901010871      NULL
Time taken: 50.215 seconds, Fetched: 1 row(s)

• ElasticSearch-Hadoop可以实现Hdfs数据与ES数据进行比对，比对计算将提交到Yarn通过Map/Reduce执行
• 更多Hive创建外部ES表属性可以参考博客和官网