mongodb数据同步到hive

背景

用户需求: 需要将 mongodb 的数据同步到 hive 表，共 2 亿+条数据，总数据量约 30G

查阅一些博客后，大致同步方法有以下几种

手动+离线

对于比较小的数据，可以先通过 mongoexport 将数据导出到本地 json 文件，再将 json 直接上传到 hdfs，创建 hive 表关联到这个文件即可

这种方式非常简单直接，但需要两次读取到写入的过程，而且如果 mongodb 中的数据很大，需要先导到本地的方式将会导致本地磁盘占用升高，不是很适合

参考: Mongoexport同步数据到Hive

工具+离线

通过 datax / seatunnel 之类的数据同步服务，数据在内存中完成同步

datax-mongodbreader

datax-hdfswriter

seatunnel-connector-mongo

seatunnel-connector-hive

实时

mongodb 在 3.6 版本支持了 change streamer，支持在客户端监听数据变更，实现实时更新

但这个同步逻辑需要依赖单独的程序去完成，这次用户需求也只要一次性的全量更新，不涉及增量更新，因此这一块就不具体看了

MongoDB Change Stream之一——上手及初体验原创

通过 datax 同步数据

任务定义

json 配置文件参考官方文档进行适配即可

注意: hdfswriter 的 defaultFS 如果指定的是集群地址，比如 hdfs://cluster_name，可能会遇到解析不了的问题，需要把 hdfs 的配置文件打入到 hdfswriter-0.0.1-SNAPSHOT.jar 包中去

参考issue

jar 打包指令参考

datax 任务定义:

{"job": {"content": [{"reader": {"name": "mongodbreader","parameter": {"address": ["host1:27017","host2:27017","host3:27017"],"collectionName": "collection_name","column": [{"index": 0,"name": "id","type": "string"},...],"dbName": "db","userName": "user","userPassword": "password"}},"writer": {"name": "hdfswriter","parameter": {"column": [{"name": "id","type": "string"}...],"compress": "SNAPPY","defaultFS": "hdfs://cluster_name","fieldDelimiter": "\u0001","fileName": "$table","fileType": "orc","path": "/user/hive/warehouse/db_name.db/table_name", // hdfs 路径需要先创建好"writeMode": "truncate"}}}],"setting": {"speed": {"channel": "2" // 任务并行度，可根据实际情况调大}}}
}

效果

启动任务

python bin/datax.py tmp/mongo_to_hive.json

等待同步完成后，查看 hdfs 对应表路径的数据，可以看到生成的 txt 文件

然后创建 hive 表关联这个路径即可

create external table table_name
(id string,...
)
COMMENT '同步表'
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' LINES TERMINATED BY '\n'
STORED AS ORC -- 和 datax 任务中的 fileType 对应
LOCATION '/user/hive/warehouse/db_name.db/table_name'; -- 和 path 对应

通过 seatunnel 同步数据

环境准备

不同于 datax，seatunnel 的安装过程稍微复杂点，而且和 hive 相关的ja包需要手动导入，因此在这里特别说明

# 下载 seatunnel 后，需要先安装插件（即各个connector）
sh bin/install-plugin.sh 2.3.3# 下载 hive 2.3.9 版本 （主要是为了拿到 jar 包，也可以从 maven 仓库单独下载下面的每个 jar 包）
wget https://downloads.apache.org/hive/hive-2.3.9/apache-hive-2.3.9-bin.tar.gz# 解压后，把以下 jar 包拷贝到 seatunnel 的 lib 下seatunnel_home=/opt/modules/seatunnelcd /opt/modules/apache-hive-2.3.9-bin/libcp hive-metastore-2.3.9.jar ${seatunnel_home}/libcp hive-common-2.3.9.jar ${seatunnel_home}/libcp hive-exec-2.3.9.jar ${seatunnel_home}/libcp libthrift-0.9.3.jar ${seatunnel_home}/libcp libfb303-0.9.3.jar ${seatunnel_home}/lib

任务定义

配置示例如下:

env {execution.parallelism = 1job.mode = "BATCH"
}source {MongoDB {uri = "mongodb://user:password@host1,host2,host3"database = "db"collection = "collection"schema = {fields {id = stringcreate_time = stringupdate_time = stringname = string...}}}}sink {Hive {table_name = "db.table"metastore_uri = "thrift://metastore_host:9083"hdfs_site_path = "/etc/hadoop/conf/hdfs-site.xml" # 和 datax 的问题类似，添加 hdfs 配置用于解析 hdfs 集群地址}
}

效果

执行任务:

./bin/seatunnel.sh --config mongo_to_hive.conf -e local"

目前还存在空值无法转换的问题，报错: Unable to convert to from nullable value null，需要等待社区解决

相关 issue: [Bug] [MongoDB-CDC] 数据中有字段值为 Null 直接异常. #5340

和 datax 的对比

配置: 都是 hive reader 和 mongo writer 的配置思路
写入方式: 两者都是把数据写入到 hdfs 路径，hive 表需要用户自己创建

seatunnel 的 hive connector 也是先从 hive 中读取 location，拿到 hdfs 的路径，参考: HiveSinkAggregatedCommitter 继承了 FileSinkAggregatedCommitter, HiveSink 继承了 BaseHdfsFileSink