如何用Logstash将约40000个嵌套JSON导入Elasticsearch索引

嘿，我来帮你搞定把这4万条嵌套JSON数据导入Elasticsearch的事儿！下面是几种靠谱的方法，你可以根据自己的技术栈和需求选最合适的：

一、提前准备：创建索引并配置映射 #

首先得先创建目标索引，并且针对你的嵌套结构和日期字段配置合适的映射——这一步很关键，能避免后续数据类型不匹配的问题。

比如在Kibana Dev Tools或者用curl发送请求：

PUT /your_index_name
{
  "mappings": {
    "properties": {
      "customerid": {"type": "keyword"},
      "date": {
        "type": "date",
        "format": "dd.MM.yyyy"
      },
      "bez": {"type": "keyword"},
      "birthdate": {
        "type": "date",
        "format": "dd.MM.yyyy"
      },
      "clientid": {"type": "keyword"},
      "address": {
        "type": "nested",  // 因为address是嵌套数组，必须设为nested类型才能正确查询内部字段
        "properties": {
          "addressid": {"type": "keyword"},
          "title": {"type": "keyword"},
          "street": {"type": "text"},
          "valid_to": {
            "type": "date",
            "format": "dd.MM.yyyy"
          },
          "valid_from": {
            "type": "date",
            "format": "dd.MM.yyyy"
          }
        }
      }
    }
  }
}

二、方法1：用Elasticsearch Bulk API（最直接） #

Bulk API是ES官方推荐的批量导入方式，适合手动处理或者简单脚本辅助的场景。

步骤1：转换JSON为Bulk格式 #

ES的Bulk API要求每两行一组：第一行是操作指令（比如index），第二行是文档内容。比如你的每条JSON要转换成这样的格式（每行一个条目）：

{"index": {"_index": "your_index_name"}}
{"customerid": "10932", "date": "16.08.2006", "bez": "xyz", "birthdate": "21.05.1990", "clientid": "2", "address": [ { "addressid": "1", "title": "Mr", "street": "main str", "valid_to": "21.05.1990", "valid_from": "21.05.1990" }, { "addressid": "2", "title": "Mr", "street": "melrose place", "valid_to": "21.05.1990", "valid_from": "21.05.1990" } ]}

如果你的原始数据是一个包含4万条对象的JSON数组，你可以用Python脚本快速转换：

import json

# 读取原始JSON文件（假设是data.json，里面是一个大数组）
with open('data.json', 'r') as f:
    docs = json.load(f)

# 生成Bulk格式的内容
with open('bulk_data.ndjson', 'w') as f:
    for doc in docs:
        # 修正你示例里的typo："tile"改成"title"
        if 'address' in doc:
            for addr in doc['address']:
                addr.pop('tile', None)  # 去掉错误的tile字段，保留title
        # 写入操作指令
        f.write(json.dumps({"index": {"_index": "your_index_name"}}) + '\n')
        # 写入文档内容
        f.write(json.dumps(doc) + '\n')

步骤2：发送Bulk请求 #

用curl发送请求到ES：

curl -X POST "http://your_es_host:9200/_bulk" -H "Content-Type: application/x-ndjson" --data-binary @bulk_data.ndjson

注意：如果数据量太大（4万条），可以把文件分成多个小文件（比如每个文件1000-5000条），避免请求超时。

三、方法2：用Logstash（适合自动化数据管道） #

如果需要定期导入或者后续有数据更新，Logstash是个省心的选择，不需要写太多脚本。

步骤1：创建Logstash配置文件（比如es_import.conf） #

input {
  file {
    path => "/path/to/your/data.json"  # 你的JSON文件路径
    start_position => "beginning"
    codec => json_lines  # 如果你的原始数据是每行一个JSON对象，用这个；如果是数组，换成json并配合split filter
  }
}

filter {
  # 修正日期格式（如果ES映射里已经配置了格式，这一步可以省略，但保险起见还是处理下）
  date {
    match => ["date", "dd.MM.yyyy"]
    target => "date"
  }
  date {
    match => ["birthdate", "dd.MM.yyyy"]
    target => "birthdate"
  }
  date {
    match => ["[address][valid_to]", "dd.MM.yyyy"]
    target => "[address][valid_to]"
  }
  date {
    match => ["[address][valid_from]", "dd.MM.yyyy"]
    target => "[address][valid_from]"
  }
  # 修正你示例里的typo：tile改成title
  mutate {
    rename => { "[address][tile]" => "[address][title]" }
  }
}

output {
  elasticsearch {
    hosts => ["http://your_es_host:9200"]
    index => "your_index_name"
    document_id => "%{customerid}"  # 可选：用customerid作为文档ID，避免重复导入
  }
  stdout { codec => rubydebug }  # 可选：打印导入日志到控制台
}

如果你的原始数据是一个大JSON数组，需要在filter里加split插件拆分数组：

filter {
  split {
    field => "message"  # 或者根据实际字段调整
  }
  # 其他filter...
}

步骤2：启动Logstash #

bin/logstash -f es_import.conf  # 根据你的Logstash安装路径调整命令

四、方法3：用Python脚本（elasticsearch-py库，灵活自定义） #

如果需要在导入前做复杂的数据清洗或者逻辑处理，用Python脚本最灵活。

步骤1：安装elasticsearch库 #

pip install elasticsearch

步骤2：编写导入脚本 #

from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk
import json

# 连接ES
es = Elasticsearch("http://your_es_host:9200")

# 读取原始数据
with open('data.json', 'r') as f:
    docs = json.load(f)

# 预处理文档：修正typo、转换日期（可选，因为映射里已经配置了格式，但手动转换更保险）
processed_docs = []
for doc in docs:
    # 修正address里的tile为title
    if 'address' in doc:
        for addr in doc['address']:
            if 'tile' in addr:
                addr['title'] = addr.pop('tile')
    # 构造Bulk需要的文档结构
    processed_docs.append({
        "_index": "your_index_name",
        "_source": doc
    })

# 批量导入
success, failed = bulk(es, processed_docs, chunk_size=1000, raise_on_error=False)
print(f"成功导入{success}条，失败{len(failed)}条")

# 打印失败的文档（如果有）
if failed:
    print("失败的文档：")
    for item in failed:
        print(item)

五、关键注意事项 #

• 日期格式：你的日期是dd.MM.yyyy格式，ES默认识别yyyy-MM-dd，所以要么在映射里配置format: dd.MM.yyyy，要么在导入前转换为标准格式。
• 批量大小：不管用哪种方法，批量大小建议设为1000-5000条，太大容易导致ES内存过载或者请求超时，太小则效率低。
• 嵌套类型：address是嵌套数组，必须在映射里设为nested类型，否则ES会把数组扁平化，导致查询内部字段时出现错误关联。
• 错误处理：导入过程中可能会有失败的文档，记得保留错误日志，后续排查后重新导入。

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