ES學習筆記-elasticsearch-hadoop導入hive數據到es的實現探究
但是有些業務也有不標準的,比如hive數據導入到ES. hive數據導入到ES, 官方組件是elasticsearch-hadoop. 其用法在前面的博客中已有介紹。 那麽其實現原理是怎樣的呢? 或者說, es-hadoop這家夥到底是怎麽把hive表的數據弄到es中去的? 為了弄清楚這個問題, 我們首先需要有一個本地的源碼環境。
s1: 下載elasticsearch-hadoop源碼。
git clone https://github.com/elastic/elasticsearch-hadoop.git
s2: 編譯源碼。直接編譯master即可。
gradlew distZip
s3: 編譯成功後,導入到intellij。 這裏註意導入build.gradle文件,就像maven項目導入pom文件一樣。
s4: 在intellij中編譯一次項目。
s5: 在本地啟動一個es, 默認的端口即可。
s6: 運行測試用例AbstractHiveSaveTest.testBasicSave()
。 直接運行是會報錯的, 需要略微修改一下代碼,添加一個類的屬性:
@Cla***ule public static ExternalResource hive = HiveSuite.hive;
如果是在windows環境下,需要新建packageorg.apache.hadoop.io.nativeio
, 然後在該package下建立NativeIO.java
類。 修改代碼如下:
// old public static boolean access(String path, Acce***ight desiredAccess) throws IOException { return access0(path, desiredAccess.acce***ight()); } // new public static boolean access(String path, Acce***ight desiredAccess) throws IOException { return true; }
這樣就運行起來了一個本地的hive到es的代碼。可以debug,了解詳細流程了。
在elasticsearch-hadoop這個比較龐大的項目中,修改代碼也比較麻煩,因此可以單獨建立一個項目hive-shgy, 然後改造這個測試類, 跑通testBasicSave()
。
由於對gradle不熟悉, 還是建立maven項目, 項目的依賴如下:
<repositories>
<repository>
<id>spring-libs</id>
<url>http://repo.spring.io/libs-milestone/</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-1.2-api</artifactId>
<version>2.6.2</version>
<scope>test</scope>
</dependency>
<dependency> <!-- 橋接:告訴Slf4j使用Log4j2 -->
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<version>2.6.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.3.6</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-cli</artifactId>
<version>1.2.1</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.2.0</version>
<scope>provided</scope>
<exclusions>
<exclusion>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch-hadoop</artifactId>
<version>6.3.0</version>
<scope>test</scope>
</dependency>
</dependencies>
這裏用到了log4j2, 所以日誌類放在前面。
接下來遷移測試代碼。遷移的原則是 若無必要,不新增類。 如果只用到了類的一個方法,那麽只遷移一個方法。
這裏的測試代碼遷移,其實就是圍繞HiveEmbeddedServer2
來構建的。個人感覺這裏比較巧妙的是,通過HiveEmbeddedServer2
啟動了一個嵌入式的hive實例。能夠執行hive sql, 而且是在一個jvm中,對於研究hive的實現原理來說,太酷了。
基礎的環境搭建好後,就可以研究elasticsearch-hadoop的源碼了, 先看源碼的結構:
elasticsearch-hadoop/hive/src/main/java/org/elasticsearch/hadoop/hive$ tree .
.
├── EsHiveInputFormat.java
├── EsHiveOutputFormat.java
├── EsSerDe.java
├── EsStorageHandler.java
├── HiveBytesArrayWritable.java
├── HiveBytesConverter.java
├── HiveConstants.java
├── HiveFieldExtractor.java
├── HiveType.java
├── HiveUtils.java
├── HiveValueReader.java
├── HiveValueWriter.java
├── HiveWritableValueWriter.java
└── package-info.java
0 directories, 14 files
這裏簡要描述一下elasticsearch-hadoop將hive數據同步到es的原理, Hive開放了StorageHandler的接口。通過StoreageHandler, 可以使用SQL將數據寫入到es,同時也可以使用SQL讀取ES中的數據。 所以, 整個es-hive, 其入口類為EsStorageHandler, 這就是整個功能的框架。 了解了EsStorageHandler後,接下來很重要的一個類就是EsSerDe, 是序列化反序列化的功能組件。它是一個橋梁,通過它實現ES數據類型和Hive數據類型的轉換。 核心類就是這兩個了。
了解了代碼的原理及結構,就可以自己仿照實現hive數據同步到mongo, hive數據同步到redis 等其他的功能了。 這樣做的好處是業務無關, 一次開發,多次使用。方便管理維護。
最後總結一下,本文沒有直接給出答案, 而是記錄了尋找答案的過程。 通過這個過程,學會將hive數據同步到其他NoSQL中,這個實踐比理解源碼更重要。
ES學習筆記-elasticsearch-hadoop導入hive數據到es的實現探究