如何把 CSV 数据导入 Elasticsearch？

最省事的两种方式：少量数据（<100MB）直接用 Kibana 的 Data Visualizer 拖拽上传；大规模或需定期导入用 Logstash 的 file input + csv filter。

方法一：Kibana 界面上传

Management → Integrations → Upload a file（或 Machine Learning → Data Visualizer）：拖入 CSV → 自动推断字段类型与分隔符 → 确认 mapping → Import 完成。适合一次性、小批量。

方法二：Logstash 管道

input {
  file {
    path => "/data/sales.csv"
    start_position => "beginning"
    sincedb_path => "/dev/null"     # 每次从头读（一次性导入场景）
  }
}
filter {
  csv {
    separator => ","
    columns => ["date", "product", "amount", "region"]
  }
  mutate {
    convert => { "amount" => "float" }    # 类型转换
  }
  date {
    match => ["date", "yyyy-MM-dd"]       # 把字符串字段变成时间戳
  }
}
output {
  elasticsearch {
    hosts => ["es:9200"]
    index => "sales-%{+YYYY.MM}"
  }
}

方法三：Python bulk 导入

import csv, json
from elasticsearch import Elasticsearch, helpers
es = Elasticsearch("http://es:9200")
with open("sales.csv") as f:
    actions = ({"_index": "sales", "_source": row} for row in csv.DictReader(f))
    helpers.bulk(es, actions)

注意事项

• CSV 全是字符串——金额、日期字段务必转换类型，否则排序/聚合出错；
• 大文件导入用 bulk 批量（每批 5-15MB）并刷新间隔调大（refresh_interval: -1 导入完再恢复）。

常见问题（FAQ）

Q：中文乱码？ 确认 CSV 是 UTF-8；GBK 文件先转码。

Q：首行表头被当数据导入了？ Logstash csv filter 加 autodetect_column_names => true 自动用表头做列名。