如何通過ES-Hadoop實現Hive讀寫Elasticsearch資料 - Elasticsearch

ES-Hadoop是Elasticsearch推出的專門用於對接Hadoop生態的工具，可以讓資料在Elasticsearch和Hadoop之間雙向移動，無縫銜接Elasticsearch與Hadoop服務，充分使用Elasticsearch的快速搜尋及Hadoop批處理能力，實現互動式資料處理。本文介紹通過ES-Hadoop組件在Hive上進行Elasticsearch資料的查詢和寫入，協助您將Elasticsearch與Hadoop生態組件結合起來，實現更靈活的資料分析。

背景資訊

Hadoop生態的優勢是處理大規模資料集，但是其缺點也很明顯，就是當用於互動式分析時，查詢時延會比較長。而Elasticsearch擅長於互動式分析，對於很多查詢類型，特別是對於Ad-hoc查詢（即席查詢），可以達到秒級。ES-Hadoop的推出提供了一種組合兩者優勢的可能性。使用ES-Hadoop，您只需要對代碼進行很小的改動，即可快速處理儲存在Elasticsearch中的資料，並且能夠享受到Elasticsearch帶來的加速效果。

ES-Hadoop的原理是將Elasticsearch作為MR、Spark或Hive等資料處理引擎的資料來源，在計算儲存分離的架構中扮演儲存的角色。這和MR、Spark或Hive的資料來源並無差異，但相對於這些資料來源，Elasticsearch具有更快的資料選擇過濾能力。這種能力正是分析引擎最為關鍵的能力之一。 ES-Hadoop架構原理圖

如果您需要瞭解ES-Hadoop與Hive更進階的配置，請參見Elasticsearch官方說明文檔。

操作流程

準備工作
建立同一專用網路下的Elasticsearch和E-MapReduce（以下簡稱EMR）執行個體、關閉Elasticsearch執行個體的自動建立索引功能並建立索引和Mapping、下載與Elasticsearch執行個體版本一致的ES-Hadoop安裝包。
步驟一：上傳ES-Hadoop JAR包至HDFS
將已下載的ES-Hadoop安裝包上傳至EMR Master節點的HDFS目錄下。
步驟二：建立Hive外表
建立Hive外表，與Elasticsearch索引中的欄位進行映射。
步驟三：通過Hive寫入索引資料
通過HiveSQL，向Elasticsearch執行個體的索引中寫入資料。
步驟四：通過Hive讀取索引資料
通過HiveSQL，讀取Elasticsearch執行個體中的索引資料。

準備工作

建立Elasticsearch執行個體。
本文使用6.7.0版本的執行個體，具體操作步驟請參見建立Elasticsearch執行個體。

關閉執行個體的自動建立索引功能，並提前建立索引和Mapping。

開啟自動建立索引功能後，可能會導致Elasticsearch自動建立的索引類型和您預期的類型不一致。例如您定義了一個欄位age，為INT類型，開啟自動建立索引後，可能將其索引成了LONG類型，因此建議手動建立索引。本文使用的索引和Mapping如下。

PUT company
{
  "mappings": {
    "_doc": {
      "properties": {
        "id": {
          "type": "long"
        },
        "name": {
          "type": "text",
          "fields": {
            "keyword": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        },
        "birth": {
          "type": "text"
        },
        "addr": {
          "type": "text"
        }
      }
    }
  },
  "settings": {
    "index": {
      "number_of_shards": "5",
      "number_of_replicas": "1"
    }
  }
}

建立與Elasticsearch執行個體在同一專用網路下的EMR叢集。
重要
Elasticsearch執行個體的私網訪問白名單預設為0.0.0.0/0，您可在安全配置頁面查看，如果未使用預設配置，您還需要在白名單中加入EMR叢集的內網IP地址：
- 請參見查看叢集列表與詳情，擷取EMR叢集的內網IP地址。
- 請參見配置執行個體公網或私網訪問白名單，配置Elasticsearch執行個體的VPC私網訪問白名單。

步驟一：上傳ES-Hadoop JAR包至HDFS

下載ES-Hadoop安裝包，其版本需要與Elasticsearch執行個體保持一致。
本文使用elasticsearch-hadoop-6.7.0.zip。
登入E-MapReduce控制台，擷取Master節點的IP地址，並通過SSH登入對應的ECS機器。
具體操作步驟請參見登入叢集。
將已下載的elasticsearch-hadoop-6.7.0.zip上傳至Master節點，並解壓獲得elasticsearch-hadoop-hive-6.7.0.jar。

建立HDFS目錄，將elasticsearch-hadoop-hive-6.7.0.jar上傳至該目錄下。

hadoop fs -mkdir /tmp/hadoop-es
hadoop fs -put elasticsearch-hadoop-6.7.0/dist/elasticsearch-hadoop-hive-6.7.0.jar /tmp/hadoop-es

步驟二：建立Hive外表

在EMR控制台的資料開發模組中，建立HiveSQL類型的作業。
具體操作步驟請參見Hive SQL作業配置。

配置作業，建立外表。

作業配置如下。

####添加jar包，僅對當前會話有效########
add jar hdfs:///tmp/hadoop-es/elasticsearch-hadoop-hive-6.7.0.jar;
####建立hive外表，與es索引進行映射#####
CREATE EXTERNAL table IF NOT EXISTS company( 
   id BIGINT,
   name STRING,
   birth STRING,
   addr STRING 
)  
STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler' 
TBLPROPERTIES(  
    'es.nodes' = 'http://es-cn-mp91kzb8m0009****.elasticsearch.aliyuncs.com',
    'es.port' = '9200',
    'es.net.ssl' = 'true', 
    'es.nodes.wan.only' = 'true', 
    'es.nodes.discovery'='false',
    'es.input.use.sliced.partitions'='false',
    'es.input.json' = 'false',
    'es.resource' = 'company/_doc',
    'es.net.http.auth.user' = 'elastic', 
    'es.net.http.auth.pass' = 'xxxxxx'
);

表 1. ES-Hadoop相關參數說明
參數	預設值	說明
es.nodes	localhost	指定Elasticsearch執行個體的訪問地址，建議使用私網地址，可在執行個體的基本資料頁面查看，詳情請參見查看執行個體的基本資料。
es.port	9200	Elasticsearch執行個體的訪問連接埠號碼。
es.net.http.auth.user	elastic	Elasticsearch執行個體的訪問使用者名稱。說明如果程式中指定elastic帳號訪問Elasticsearch服務，後續在修改elastic帳號對應密碼後需要一些時間來生效，在密碼生效期間會影響服務訪問，因此不建議通過elastic來訪問。建議在Kibana控制台中建立一個符合預期的Role角色使用者進行訪問，詳情請參見通過Elasticsearch X-Pack角色管理實現使用者權限管控。
es.net.http.auth.pass	/	Elasticsearch執行個體的訪問密碼。
es.nodes.wan.only	false	開啟Elasticsearch叢集在雲上使用虛擬IP進行串連，是否進行節點嗅探： true：設定 false：不設定
es.nodes.discovery	true	是否禁用節點發現： true：禁用 false：不禁用重要使用Elasticsearch，必須將此參數設定為false。
es.input.use.sliced.partitions	true	是否使用slice分區： true：使用。設定為true，可能會導致索引在預讀階段的時間明顯變長，有時會遠遠超出查詢資料所耗費的時間。建議設定為false，以提高查詢效率。 false：不使用。
es.index.auto.create	true	通過Hadoop組件向Elasticsearch叢集寫入資料，是否自動建立不存在的index： true：自動建立 false：不會自動建立
es.resource	/	指定要讀寫的index和type。
es.mapping.names	/	表欄位與Elasticsearch的索引欄位名映射。
es.read.metadata	false	操作Elasticsearch欄位涉及到_id之類的內部欄位，請開啟此屬性。

更多的ES-Hadoop配置項說明，請參見官方配置說明。

儲存並運行作業。
運行成功後，結果如下。

步驟三：通過Hive寫入索引資料

建立一個HiveSQL類型的寫資料作業。

作業配置如下。

add jar hdfs:///tmp/hadoop-es/elasticsearch-hadoop-hive-6.7.0.jar;
INSERT INTO TABLE company VALUES (1, "zhangsan", "1990-01-01","No.969, wenyixi Rd, yuhang, hangzhou");
INSERT INTO TABLE company VALUES (2, "lisi", "1991-01-01", "No.556, xixi Rd, xihu, hangzhou");
INSERT INTO TABLE company VALUES (3, "wangwu", "1992-01-01", "No.699 wangshang Rd, binjiang, hangzhou");

儲存並運行作業。
運行成功後，登入Elasticsearch執行個體的Kibana控制台，查看company索引資料。
登入Kibana控制台的具體操作步驟，請參見登入Kibana控制台。您可以在Kibana控制台中，執行以下命令查看company索引資料。
```
GET company/_search
```
執行成功後，返回結果如下。

步驟四：通過Hive讀取索引資料

建立一個HiveSQL類型的讀資料作業。

作業配置如下。

add jar hdfs:///tmp/hadoop-es/elasticsearch-hadoop-hive-6.7.0.jar;
select * from company;

儲存並運行作業。

常見問題

Q：Hive讀寫Elasticsearch資料時，遇到如下報錯資訊，該如何處理？

報錯資訊： FAILED: Execution Error, return code -101 from org.apache.hadoop.hive.ql.exec.mr.MapRedTask. Could not initialize class org.elasticsearch.hadoop.rest.commonshttp.CommonsHttpTransport。

A：報錯的原因是EMR 5.6.0版本的Hive組件缺少commons-httpclient-3.1.jar檔案。處理方法為手動增加缺少的檔案到Hive的lib目錄下，檔案下載地址，請參見commons-httpclient-3.1。