本文csdn地址

背景

用户需求: 需要将 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 任务定义:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
{
  "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" // 任务并行度，可根据实际情况调大
      }
    }
  }
}

效果

启动任务

1
python bin/datax.py tmp/mongo_to_hive.json

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

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

1
2
3
4
5
6
7
8
9
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包需要手动导入，因此在这里特别说明

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 下载 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/seatunnel

cd /opt/modules/apache-hive-2.3.9-bin/lib

cp hive-metastore-2.3.9.jar ${seatunnel_home}/lib

cp hive-common-2.3.9.jar ${seatunnel_home}/lib

cp hive-exec-2.3.9.jar ${seatunnel_home}/lib

cp libthrift-0.9.3.jar ${seatunnel_home}/lib

cp libfb303-0.9.3.jar ${seatunnel_home}/lib

任务定义

配置示例如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
env {
  execution.parallelism = 1
  job.mode = "BATCH"
}

source {
  MongoDB {
    uri = "mongodb://user:password@host1,host2,host3"
    database = "db"
    collection = "collection"
    schema = {
      fields {
            id = string
            create_time = string
            update_time = string
            name = 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 集群地址
  }
}

效果

执行任务:

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

目前还存在 空值无法转换的问题，报错: Unable to convert to <STRING> 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

  • 运行环境: datax 只能本地运行，seatunnel 支持 local、flink、spark 等运行模式