如何把 CSV 数据导入 Elasticsearch？

最省事的两种方式：少量数据（上传上限依 Kibana 版本和部署配置核对）直接用 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 => "/var/lib/logstash/sincedb-sales"  # 持久可写位置，防重启从头读
  }
}
filter {
  csv {
    separator => ","
    columns => ["date", "product", "amount", "region"]
    skip_header => true
  }
  mutate {
    convert => { "amount" => "float" }    # 类型转换
  }
  date {
    match => ["date", "yyyy-MM-dd"]       # 把字符串字段变成时间戳
  }
}
output {
  elasticsearch {
    hosts => ["es:9200"]
    ilm_enabled => false
    data_stream => false
    index => "sales-%{+YYYY.MM}"
  }
}

skip_header 要求 pipeline.workers: 1，且表头与配置列名匹配；按行读取不适合含引号内换行的多行 CSV。date filter 默认更新 @timestamp，不会自动改变 date 字段自身，需预设合适 mapping。

方法三：Python bulk 导入

import csv, json
from elasticsearch import Elasticsearch, helpers
es = Elasticsearch("http://es:9200")
def actions_from_csv(f):
    for row in csv.DictReader(f):
        row["amount"] = float(row["amount"])
        yield {"_index": "sales", "_source": row}

with open("sales.csv", newline="", encoding="utf-8") as f:
    actions = actions_from_csv(f)
    helpers.bulk(es, actions)

注意事项

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

常见问题（FAQ）

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

Q：首行表头被当数据导入了？ 可用 skip_header + 明确 columns，或 autodetect_column_names 配合相应表头跳过策略；这些表头检测选项要求单 worker。bulk 导入还需处理逐项失败及重试去重，不能只看 HTTP 成功。