Python读取CSV逐行插入Elasticsearch遇IOError报错求助

嘿，我来帮你梳理下问题所在，顺便给你修正后的代码～

首先看你遇到的IOError: File not open for writing错误，根源有两个：

1. 你误用了json.dump()——这个函数是用来把数据写入文件的，但你现在需要的是把CSV行数据转成Elasticsearch能接受的格式，应该用json.dumps()（注意是dumps，带s）；
2. 你打开文件用的是默认的只读模式（r），但json.dump()试图往这个文件里写内容，自然就报错了。

另外还有个隐藏问题：你在循环里写了[row for row in reader]，这会一次性把CSV的所有剩余行读完，导致你的外层for row in reader循环只执行一次就结束了，根本不会逐行插入数据。

还有个小建议：用csv.reader得到的是纯列表，直接存到ES里没有字段名，可读性很差，不如用csv.DictReader，它会自动把CSV的第一行作为字段名，把每行转成键值对字典，这样存在ES里的数据结构更合理。

给你修正后的代码：

import csv
from elasticsearch import Elasticsearch

# 连接Elasticsearch，测试连接是否成功
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])
print("Elasticsearch连接状态:", es.ping())

def csv_to_es(file_path):
    # 用DictReader读取CSV，自动映射字段名
    with open(file_path, 'r') as f_obj:
        reader = csv.DictReader(f_obj)
        doc_id = 1
        for row in reader:
            # 注意：如果你的ES是7.x及以上版本，要删掉doc_type参数，这个属性已经被移除了
            response = es.index(
                index='product',
                doc_type='prod',
                id=doc_id,
                body=row  # ES客户端会自动把字典转成JSON，不用手动处理
            )
            print(f"插入第{doc_id}条数据结果: {response['result']}")
            doc_id += 1

if __name__ == "__main__":
    csv_file = "/home/Documents/csv/acsv.csv"
    csv_to_es(csv_file)

关键修改点说明： #

• 用csv.DictReader替代csv.reader：比如你的CSV第一行是商品名,价格,库存，那么每行数据会变成{'商品名': 'xxx', '价格': '99', '库存': '10'}，存到ES里的数据结构清晰，方便后续查询。
• 移除错误的json.dump()：es.index的body参数直接接受Python字典，客户端会自动序列化JSON，不需要手动转换。如果你一定要手动转，用json.dumps(row)而不是json.dump()。
• 修复循环逻辑：直接遍历reader的每一行，不再重复读取整个reader，保证逐行插入。
• 增加连接测试：用es.ping()快速验证是否能连上ES服务，避免因为连接失败白忙活。

额外注意事项： #

1. 如果你的Elasticsearch是7.x及以上版本，一定要删掉doc_type='prod'这个参数，ES从7.x开始已经移除了文档类型的概念，保留的话会报错。
2. 如果CSV里有数值型字段（比如价格、库存），建议把字符串转成对应类型，比如row['价格'] = float(row['价格'])，这样ES能正确识别字段类型，方便后续做数值筛选、聚合等操作。
3. 如果你的CSV数据量很大，逐行插入效率很低，建议用elasticsearch.helpers.bulk()做批量插入，能大幅提升速度。

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