Python Elasticsearch Bulk API异常：仅导入5000条数据后中断

嘿，这个问题我之前帮人排查过，本质就是编码不匹配在搞事情！你的MySQL数据里肯定藏着非ASCII字符（比如带重音的字母、特殊符号这类），但Python默认用ASCII编码去处理这些字符，自然就解码失败中断了。咱们一步步来搞定它：

1. 先把MySQL连接的编码拉对 #

这是最核心的一步！你的get_connection()函数里，连接MySQL时必须指定支持全Unicode的字符集utf8mb4（别用普通的utf8，它不支持emoji和一些特殊字符）。比如用pymysql的话，修改后的连接函数应该是这样：

def get_connection():
    import pymysql
    return pymysql.connect(
        host="你的数据库地址",
        user="你的用户名",
        password="你的密码",
        db="你的数据库名",
        charset="utf8mb4",  # 关键参数！确保数据以UTF-8格式取出
        cursorclass=pymysql.cursors.DictCursor  # 可选，返回字典格式的结果，处理起来更方便
    )

如果用的是mysqlclient（MySQLdb的替代分支），参数设置是一样的，只要加上charset='utf8mb4'就行。这一步能保证从数据库捞出来的字符串本身就是正确的Python3 str类型，而不是乱码的字节串。

2. 显式处理字节串（如果还有问题） #

要是指定编码后还是报错，可能某些字段返回的是bytes类型（比如旧版驱动的问题），这时候就得手动转成字符串：

cursor.execute("SELECT id, name, description FROM products WHERE catid = %s", (catid,))
for row in cursor.fetchall():
    processed_row = {}
    for key, value in row.items():
        # 把字节串用UTF-8解码成字符串
        if isinstance(value, bytes):
            processed_row[key] = value.decode("utf-8")
        else:
            processed_row[key] = value
    # 把处理后的行加入Bulk请求队列

3. 分批次处理，别一次性扛10000条 #

一次性加载10000条数据不仅容易触发编码问题，还可能搞出内存溢出。改成小批量处理（比如每次1000条），既能降低出错概率，还能快速定位到出问题的批次：

from elasticsearch.helpers import bulk

batch_size = 1000
cursor.execute("SELECT id, name, description FROM products WHERE catid = %s", (catid,))

while True:
    # 每次取一批数据
    rows = cursor.fetchmany(batch_size)
    if not rows:
        break  # 取完所有数据就退出循环
    
    actions = []
    for row in rows:
        # 先处理编码
        processed_row = {k: v.decode("utf-8") if isinstance(v, bytes) else v for k, v in row.items()}
        # 构建Bulk需要的动作格式
        actions.append({
            "_index": "你的ES索引名",
            "_id": processed_row["id"],
            "_source": processed_row
        })
    
    # 执行批量导入
    try:
        bulk(es, actions)
        print(f"成功导入{len(actions)}条数据")
    except Exception as e:
        print(f"当前批次导入失败: {str(e)}")
        # 打印出错批次的第一条数据，方便排查
        print("出错批次的第一条记录:", rows[0])

4. 精准定位出错的那条记录 #

如果还是搞不定，就给单条记录加个异常捕获，直接找到哪个“捣蛋鬼”导致的错误：

actions = []
for row in cursor.fetchall():
    try:
        processed_row = {k: v.decode("utf-8") if isinstance(v, bytes) else v for k, v in row.items()}
        actions.append({
            "_index": "你的ES索引名",
            "_id": processed_row["id"],
            "_source": processed_row
        })
    except UnicodeDecodeError as e:
        print(f"处理记录ID={row['id']}时出错: {str(e)}")
        print("原始数据内容:", row)
        # 可以选择跳过这条记录，或者手动修复后再导入
        continue

# 最后执行批量导入
bulk(es, actions)

按照这几步来，应该就能顺利把所有10000条数据导入Elasticsearch了！

内容的提问来源于stack exchange，提问作者Irfan Ali