使用Python将MongoDB数据同步到Elasticsearch

使用Python将MongoDB数据同步到Elasticsearch

版本说明&＃xff1a;Python 3.7 PyMongo&＃xff1a;3.11.0 Elasticsearch:5.5.3
话不多说直接鲁码&＃xff08;如遇到什么问题欢迎留言讨论&＃xff09;

# coding:utf8
# 将mongodb中的数据同步到Es中
from pymongo import MongoClient
from elasticsearch import Elasticsearch, helpers
import json
import logging# mongodb 数据库地址
CONN_ADDR1 &＃61; &＃39;更换为自己的MongoDB地址&＃39;
USERNAME &＃61; &＃39;MongoDB用户名&＃39;
PASSWORD &＃61; &＃39;MongoDB密码&＃39;
DB &＃61; "MongoDB的库"
COLLECTION &＃61; "MongoDB集合"# Es 数据库地址
class ElasticObj:def __init__(self, index_name, index_type, ip):""":param index_name: 索引名称:param index_type: 索引类型"""self.index_name &＃61; index_nameself.index_type &＃61; index_type# 无用户名密码状态# self.es &＃61; Elasticsearch([ip])# 用户名密码状态# 连接ESself.es &＃61; Elasticsearch([ip], http_auth&＃61;(&＃39;ES用户名&＃39;, &＃39;ES密码&＃39;), port&＃61;9200)# def chaxun(self):#查询所有数据# db &＃61; self.client[&＃39;xcc_company_name&＃39;]# collection &＃61; db[&＃39;name_A&＃39;]# data_qiyes &＃61; collection.find({}, no_cursor_timeout&＃61;True)# return data_qiyes# 创建索引def create_index(self):&＃39;&＃39;&＃39;创建索引,创建索引名称为ott&＃xff0c;类型为ott_type的索引:param ex: Elasticsearch对象:return:&＃39;&＃39;&＃39;# 创建映射_index_mappings &＃61; {"mappings": {self.index_type: {"properties": {"name": {&＃39;type&＃39;: &＃39;text&＃39;},"password": {&＃39;type&＃39;: &＃39;text&＃39;},"birthplace": {&＃39;type&＃39;: &＃39;text&＃39;}}}}}if self.es.indices.exists(index&＃61;self.index_name) is not True:res &＃61; self.es.indices.create(index&＃61;self.index_name, body&＃61;_index_mappings, ignore&＃61;400)print(res)# 打印详细的日志def detailedlog(self):logging.basicConfig(format&＃61;&＃39;%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s&＃39;,level&＃61;logging.DEBUG)logging.debug(&＃39;debug 信息&＃39;)logging.info(&＃39;info 信息&＃39;)logging.warning(&＃39;warning 信息&＃39;)logging.error(&＃39;error 信息&＃39;)logging.critical(&＃39;critial 信息&＃39;)# 插入数据def insert_data(self, p&＃61;0):ACTIONS &＃61; []p &＃61; 1bulk_num &＃61; 200# 连接mongo库client &＃61; MongoClient(CONN_ADDR1, port&＃61;27017)client.admin.authenticate(USERNAME, PASSWORD)mydb &＃61; client[DB]mycol &＃61; mydb[COLLECTION]# find()最大限制是101条 使用mycol.find({}, no_cursor_timeout&＃61;True) 查询所有Cursor &＃61; mycol.find({}, no_cursor_timeout&＃61;True)n &＃61; 0for mongoRecordRes in Cursor:data &＃61; []# for mongoRecordRes in list(mycol.find({}, no_cursor_timeout&＃61;True)):n &＃61; n &＃43; 1print("开始处理数据" &＃43; str(n) &＃43; "条")data_value &＃61; []data_name &＃61; ["corporateAnnualInfoId", "companyId", "nbYear", "annualCount", "basicInfo", "onlineStoreInfo","shareContributive", "frimAssetInfo", "socialInfo"]。。。&＃xff08;这里编写自己处理数据的逻辑&＃xff09;# print(data_value)# print(data_name)data_name_value &＃61; dict(zip(data_name, data_value))data.append(data_name_value)# print(data)# 遍历data数据for list_line in data:# 去掉引号# list_line &＃61; eval(list_line)# print("-" * 10)# print(list_line)# print("-" * 10)action &＃61; {"_index": self.index_name,"_type": self.index_type,"_id": list_line["corporateAnnualInfoId"], # _id 也可以默认生成&＃xff0c;不赋值"_source": {"corporateAnnualInfoId": list_line["corporateAnnualInfoId"],"companyId": list_line["companyId"],"nbYear": list_line["nbYear"],"annualCount": list_line["annualCount"],"basicInfo": list_line["basicInfo"],"onlineStoreInfo": list_line["onlineStoreInfo"],"shareContributive": list_line["shareContributive"],"frimAssetInfo": list_line["frimAssetInfo"],"socialInfo": list_line["socialInfo"]}}p &＃43;&＃61; 1ACTIONS.append(action)print(ACTIONS)# success, _ &＃61; helpers.bulk(self.es, action, index&＃61;self.index_name, raise_on_error&＃61;True)# 批量处理if len(ACTIONS) &＃61;&＃61; bulk_num:print(&＃39;插入&＃39;, p / bulk_num, &＃39;批数据&＃39;)print(len(ACTIONS))success, _ &＃61; helpers.bulk(self.es, ACTIONS, index&＃61;self.index_name, raise_on_error&＃61;True)del ACTIONS[0:len(ACTIONS)]print(success)if len(ACTIONS) > 0:success, _ &＃61; helpers.bulk(self.es, ACTIONS, index&＃61;self.index_name, raise_on_error&＃61;True)del ACTIONS[0:len(ACTIONS)]print(&＃39;Performed %d actions&＃39; % success)if __name__ &＃61;&＃61; &＃39;__main__&＃39;:#这里IP更换为自己的obj &＃61; ElasticObj("IndexName", "IndexType",ip&＃61;"es-cn-********.com")#创建索引 本人这里没有使用只是测试了一下#obj.create_index()obj.detailedlog()#执行数据的插入obj.insert_data()