python elasticsearch 批量写入数据

Posted 2021-02-02 ❦火鸟网络

from elasticsearch import Elasticsearch
from elasticsearch import helpers
import pymysql
import time

# 连接ES
es = Elasticsearch(
    [‘127.0.0.1‘],
    port=9200
)

# 连接MySQL
print("Connect to mysql...")
mysql_db = "test"
m_conn = pymysql.connect(‘localhost‘, ‘root‘, ‘数据库密码‘, ‘table表‘)
m_cursor = m_conn.cursor()

try:
    num_id = 0
    while True:
        s = time.time()
        # 查询数据
        sql = "select good_id, title,description from goods LIMIT {}, 100000".format(num_id*100000)
        # 这里假设查询出来的结果为 张三 26 北京
        m_cursor.execute(sql)
        query_results = m_cursor.fetchall()

        if not query_results:
            print("MySQL查询结果为空 num_id=<{}>".format(num_id))
            break
        else:
            actions = []
            for line in query_results:
            # 拼接插入数据结构
                action = {
                    "_index": "tenco2019",
                    "_type": "goods",
                    "_id":line[0],
                    "_source": {
                        "good_title": line[1],
                        "good_description": line[2],
                    }
                }
                # 形成一个长度与查询结果数量相等的列表
                actions.append(action)
            # 批量插入
            a = helpers.bulk(es, actions)
            e = time.time()
            print("{} {}s".format(a, e-s))
        num_id += 1

finally:
    m_cursor.close()
    m_conn.close()
    print("MySQL connection close...")