同步 Hive 数据仓库数据到 Elasticsearch 的工具更新

同步 hive 数据到 Elasticsearch 的工具

可选全量（默认）和增量；同时支持编写 SQL 产生中间结果表，再导入到 ES ；

已经支持从 impala 渠道导数据，极大提升导数据速度

采用分页查询机制，数据集过多时不会撑爆内存；

我实习期的公司的数据分析、产品、运营经常需要看各种报表，多是分析统计类需求，Elasticsearch 适合做统计分析，结合 Kibana 可以直接生成报表！对这类常有的统计类需求，我的通常做法是从 hive 数据仓库导数据表到 ES，或者先用 HQL 或 ImpalaSQL 筛选出结果表，ES 拿到数据再进行聚合统计，如(Date Histogram)每天、每周、每月、某人的数据。

kibana 再生成各类可视化图表，最终数据直观展现！

Elastic 官方已经有了 Hive integration 的同步工具，但是由于使用的 hive 版本太低，ES 又已经是最新版本，尝试使用 hive integration 时一直报错，为尽快适应当前需求手动造了该轮子。

力求简洁的配置，方便使用。

脚本使用说明

环境: Python2 Python3
命令 #python hive_to_es.py config=<配置文件路径>

配置文件使用说明：使用.ini 后缀的配置文件

;Elasticsearch 地址(有多节点，地址用逗号','隔开)、用户名、密码
[es]
hosts = 192.168.3.100:9200
username = elastic
password = 888888

;存入的 es 的 index 默认等于 hive 或 impala 中的数据库名称
;在这里可配置自定义全局 index 名，所有导出表将默认导到该 index
;default_index = tqc_ttt

;数据平台，默认是 hive
;by = impala

;Hive 地址、端口、数据库名、用户等配置
[hive]
host = 127.0.0.1
port = 10000
user = hiveuser
auth_mechanism = PLAIN
database = dbname

;Impala 地址、端口、数据库名等配置
[impala]
host = 127.0.0.1
port = 21050
database = dbname


;需要导到 ES 的各个表的名称，同时也是导到 ES 的 type 名(可配置)；
;如果是通过 SQL 筛选出新的结果表再导入 ES，结果表名称可自定义，但必须再在下面给出 SQL 文件路径的配置
[table]
tables = student,score,teacher,my_result_a,my_result_b

;SQL 筛选结果表 my_result_a
[my_result_a]
;通过编写 HQL 或 ImpalaSQL 获得新的结果集表导入 ES 时的 SQL 文件路径，目前还不支持带有注释的 SQL
sql_path = ./sql/hql_test1.sql

;再定义另一想要导出到 ES 的结果表
[my_result_b]
sql_path = ./sql/hql_test2.sql


# 如需要对导出表或者结果表作出更多配置，可进行如下可选配置

;配置头为对应要导出的表或结果表的名称
;[student]

;若不使用默认 index，则配置此目标 index
;es_index = tqc_test
;若不使用默认 type，则配置此目标 type ；默认 type 与表名一致
;es_type = tqc_test_type

;限定导出的字段
;columns = date,name,age,address,sex

;字段名映射，这里 hive 表中的 name 字段映射为 ES 中的 name_in_es，sex 字段映射为 ES 中的 sex_in_es...
;column_mapping = date=@timestamp,name=name_in_es,sex=sex_in_es

;where 条件语句，导表时限定字段数据值条件
;where = age>20 AND name LIKE 'abc%'

;通过编写 HQL 或 ImpalaSQL 获得新的结果集表导入 ES 时的 SQL 文件路径，目前还不支持带有注释的 SQL
;sql_path = ./sql/hql_test1.sql

;分页查询配置，为了防止一次查询出所有数据，导致结果集过大，内存吃不消，无分页配置时默认分页大小 30000
;page_size = 1000

;全量 & 增量：导入数据前是否清空该 type 下所有数据，默认=true：清空原有 type 中数据，再把新数据导入 ES （全量更新数据）。
;overwrite = false