使用Elasticsearch Bulk API批量导入数据时遇UnicodeDecodeError

你的问题根源很明确：文档ID的生成格式不兼容Bulk API的序列化流程。

你用hashlib.md5(...).digest()生成ID时，这个方法返回的是原始字节串（比如错误日志里的'8\x1dI\xa2\xe9\xa2H-\xa6\x0f\xbd=\xa7CY\xa3'），而Elasticsearch的Bulk序列化器在把整个action字典转成JSON时，会尝试用UTF-8解码这个字节串，但这些原始字节里包含了不符合UTF-8规则的字符，直接触发了UnicodeDecodeError。

而Index API能成功，是因为它的id参数可以直接接受字节串，不需要经过完整的JSON序列化流程——但代价就是导入速度慢到无法接受。

最简单的解决方案：改用hexdigest()生成ID #

把digest()换成hexdigest()，它会返回一个十六进制字符串（比如'81d49a2e9a2482da60fbd3da74357a3'），完全符合UTF-8编码规范，序列化时不会有任何问题。

修改后的核心代码片段：

def set_data(input_file):
    with open(input_file) as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            sendtime = datetime.datetime.strptime(row['sendTime'].split('.')[0], ORIGINAL_FORMAT)
            yield {
                "_index": '{0}-{1}_{2}'.format(
                    INDEX_PREFIX,
                    sendtime.replace(day=1).strftime(INDEX_DATE_FORMAT),
                    (sendtime.replace(day=1) + relativedelta(months=1)).strftime(INDEX_DATE_FORMAT)),
                "_type": 'data',
                # 替换为hexdigest()生成字符串格式的ID
                '_id': hashlib.md5("{0}{1}{2}{3}{4}".format(sendtime, row['IMSI'], row['MSISDN'], int(row['ruleRef']), int(row['sponsorRef']))).hexdigest(),
                "_source": {
                    'body': {
                        'status': int(row['status']),
                        'sendTime': sendtime
                    }
                }
            }

额外优化建议 #

1. 固定时间戳格式：sendtime是datetime对象，直接参与字符串格式化可能会因为环境差异导致ID不一致，建议转成固定格式的字符串，比如sendtime.strftime(ORIGINAL_FORMAT)。
2. 缩小索引删除范围：你的代码里es.indices.delete(index='*')会删除所有索引，风险极高，建议改成es.indices.delete(index=f"{INDEX_PREFIX}-*")，只删除目标前缀的索引。

修改后，Bulk API就能正常处理你的批量导入请求，同时保持它应有的速度优势，不会再出现编码错误了。

内容的提问来源于stack exchange，提问作者Zeinab Abbasimazar