mongodb数据如何导入到clickhouse

一 背景说明

1 开发要求mongodb里的数据需要导入到clickhouse，方便他们分析，因此才有了如下的操作， 刚开始找了很多第三方的数据迁移软件，比如tapdata有这个功能，不过用过几次，经常报错，并且也是收费的。因此才决定自己写python脚本解决这个问题。

2 数据能否顺利导入，发现跟创建ck库里面表的字段类型有着密切的关系

二 在ck里需要提前创建好指定的表

下面是我在ck里面创建了一个core_customer_test 这个表，像create_time,update_time，应该都是Datetime的类型，我这里选择了string,不然我的导入脚本就会报错

CREATE TABLE mongodb.core_customer_test
(
    `create_time` Nullable(String),
    `dep_id`  Nullable(String),
    `is_delete` Nullable(String),
    `invitor_userid` Nullable(String),
    `name` Nullable(String),
    `qx_name` Nullable(String),
    `unionid` Nullable(String),
    `video_add_time` Nullable(String),
    `lunar_birthday` Nullable(String),
    `external_userid` Nullable(String),
    `follow_user` Nullable(String),
    `_id` Varchar,
    `birthday` Nullable(String),
    `update_time` Nullable(String),
    `sex` Nullable(String),
)
ENGINE = MergeTree

ORDER BY (_id);

三 数据同步脚本

3.1 全量同步脚本

import pymongo
import pymysql
import sys
from clickhouse_driver import Client
#import datetime
from datetime import datetime
from datetime import datetime, timedelta
#dd = collection.find().limit(2)
mydb = Client(
  host="clickhouse_ip.clickhouse.ads.aliyuncs.com",
  user="root_db",
  password="xxxxxx",
  database="mongodb"
)  #clickhouse
#mycursor = mydb.cursor()
client = pymongo.MongoClient('mongodb://root:xxxxxxx..@172.19.144.111:27018/')
db = client['sjzx']
collection = db['core_customer']
skip_num=0
count=collection.count_documents({})
print(count)
read_num=0
clickhouse_data = []

while True:
    clickhouse_data = []
    #print(skip_num)
    if read_num * 100 >= count:
       break
    dd = collection.find().limit(200).skip(skip_num)
    for row in dd:

        print(clickhouse_data)
        clickhouse_data = []
        converted_row = [
            str(row["_id"]),  # 转换为字符串类型
            str(row.get("birthday",'')),  # 转换为整数类型
            #row['create_time'].replace(microsecond=0),
            str(row.get("create_time",'')),
            str(row.get("dep_id",'')),  # 转换为浮点数类型
            #row.get['dep_id',''],
            str(row.get("is_delete",'')),  # 转换为浮点数类型
            str(row.get("name",'')),  # 转换为浮点数类型
            str(row.get("qx_name",'')),  # 转换为浮点数类型
            str(row.get("sex",'')),  # 转换为浮点数类型
            #row['update_time'].replace(microsecond=0),
            str(row.get("update_time",'')),  # 转换为浮点数类型
            str(row.get("video_add_time",'')),  # 转换为浮点数类型
            str(row.get("lunar_birthday",'')),  # 转换为浮点数类型
            str(row.get("invitor_userid",'')),  # 转换为浮点数类型
            str(row.get("external_userid",'')),  # 转换为浮点数类型
            str(row.get("follow_user",'')),  # 转换为浮点数类型
        ]

        clickhouse_data.append(converted_row)

        insert_query = ('INSERT INTO core_customer_test (_id, birthday, create_time,dep_id,is_delete,name,qx_name,sex,update_time,video_add_time,lunar_birthday, invitor_userid, external_userid, follow_user) VALUES')
        mydb.execute(insert_query, clickhouse_data,types_check=True)
    skip_num = skip_num + 200
    read_num = read_num + 1
mydb.disconnect()

3.2 增量同步脚本

增量同步是暂定第二天的1点钟，同步前一天的所有数据，在定时任务计划里固定时间运行 比如：2023-4-13 的1点钟，会同步 2023-4-12的一天的数据，以此类推。

import pymongo
import pymysql
import sys
from clickhouse_driver import Client
#import datetime
from datetime import datetime
from datetime import datetime, timedelta
#dd = collection.find().limit(2)
mydb = Client(
  host="clickhouse_ip.clickhouse.ads.aliyuncs.com",
  user="root_db",
  password="xxxxxxx",
  database="mongodb"
)  #clickhouse
#mycursor = mydb.cursor()
client = pymongo.MongoClient('mongodb://root:xxxxx..@172.19.144.111:27018/')
db = client['sjzx']
collection = db['core_customer']
skip_num=0
yes_date = datetime.now() - timedelta(days=1)

yes_date_start = yes_date.strptime(yes_date.strftime('%Y-%m-%d') + ' 00:00:00', '%Y-%m-%d %H:%M:%S')
yes_date_end = yes_date.strptime(yes_date.strftime('%Y-%m-%d') + ' 23:59:59', '%Y-%m-%d %H:%M:%S')
myquery = {'create_time': {'$gte': yes_date_start, '$lte': yes_date_end}}
count=collection.count_documents({})
print(count)
read_num=0
clickhouse_data = []

while True:
    clickhouse_data = []
    #print(skip_num)
    if read_num * 100 >= count:
       break
    dd = collection.find(myquery).limit(200).skip(skip_num)
    for row in dd:

        print(clickhouse_data)
        clickhouse_data = []
        converted_row = [
            str(row["_id"]),  # 转换为字符串类型
            str(row.get("birthday",'')),  # 转换为整数类型
            str(row.get("create_time",'')),  # 转换为整数类型
            str(row.get("dep_id",'')),  # 转换为浮点数类型
            str(row.get("is_delete",'')),  # 转换为浮点数类型
            str(row.get("name",'')),  # 转换为浮点数类型
            str(row.get("qx_name",'')),  # 转换为浮点数类型
            str(row.get("sex",'')),  # 转换为浮点数类型
            str(row.get("update_time",'')),  # 转换为浮点数类型
            str(row.get("video_add_time",'')),  # 转换为浮点数类型
            #row['video_add_time'].replace(microsecond=0),
            str(row.get("lunar_birthday",'')),  # 转换为浮点数类型
            str(row.get("invitor_userid",'')),  # 转换为浮点数类型
            str(row.get("external_userid",'')),  # 转换为浮点数类型
            str(row.get("follow_user",'')),  # 转换为浮点数类型
        ]

        clickhouse_data.append(converted_row)

        insert_query = ('INSERT INTO core_customer_test (_id, birthday, create_time,dep_id,is_delete,name,qx_name,sex,update_time,video_add_time,lunar_birthday, invitor_userid, external_userid, follow_user) VALUES')
        mydb.execute(insert_query, clickhouse_data,types_check=True)
    skip_num = skip_num + 200
    read_num = read_num + 1
mydb.disconnect()
#mydb.close()

四 到clickkhosue查看数据

最后，要感谢郑照辉的帮助。