欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

python 批量导入mysql数据到Elastic Search

程序员文章站 2022-06-11 10:44:30
...

实现功能:批量导入mysql数据到Elastic Search

软件版本:mysql5.8,Elastic Search7.4,python3.6

代码:

# -*- coding: utf-8 -*-

import pymysql
def get_local_mysql():
    try:
        mysql_conn = pymysql.connect(
            host='127.0.0.1',
            port=3306,
            user='root',
            password='123456',
            db='demo',
            charset='utf8'
        )
        return mysql_conn
    except Exception as e:
        print(e)

import datetime
import re
import time


import os
import traceback
from elasticsearch import helpers
from elasticsearch import Elasticsearch
es = Elasticsearch(hosts="192.168.184.111:9200")

settings = {
    "elsticsearch_nodes": [
        '192.168.184.111:9200'
    ]
}
es_client = None
region_dict = {}
es_dict = {}

def get_es_client():
    pid = os.getpid()
    if pid in es_dict:
        return es_dict[pid]
    else:
        es_client = Elasticsearch(
                settings['my-es'],
                sniff_on_start = True,
                sniff_on_connection_fail = True,
                sniffer_timeout = 60,
                maxsize = 10
            )
        es_dict[pid] = es_client
        return es_client

def es_main():
    
    mysql_conn = get_local_mysql()
    
    cursor0 = mysql_conn.cursor()
    cursor0.execute(
        '''
            SELECT
                *
            FROM
                tb_demo t
            ORDER BY
                t.id
            limit 
                4000000, 1000000
        '''#.format(i) 
    )
    res0 = cursor0.fetchall()
    
    list1 = []
    for row in cursor0.description:
        list1.append(row[0])
    tuple(list1)
    
    actions = []
    tb_name = "tb_demo"
    cnt = 0
    for r in res0:
        cnt += 1
        
        data = dict(zip(list1, r))
        
        print(cnt)
        
        try:                                                
            action = {"_index":tb_name, '_id': data["id"], "_source":data} 
            actions.append(action)
               
            # 每1000个一组批量导入
            if len(actions) == 1000:
                print("execute 1000 action")
                success,errors = helpers.bulk(es, actions, raise_on_error = True)
                actions = []
        except:        
            print('traceback.format_exc():\n%s' % traceback.format_exc())
    if len(actions) > 0:
        try:
            success,errors = helpers.bulk(es, actions, raise_on_error = True)
        except Exception as e:
            print(e)

if __name__ == '__main__':
    es_main()