Elasticsearch _bulk操作中能否生成字段组合哈希的自定义_id？

你的需求可以实现，但不能直接在Bulk请求的元数据中写动态哈希表达式——Elasticsearch的Bulk API不支持在请求体里执行这种动态计算逻辑，你示例里的${hash(doc['@timestamp'] + ...)}语法是不被识别的。

下面是两种可行的实现方式：

1. 客户端提前计算ID再发送Bulk请求 #

在发送Bulk请求前，在业务代码中手动拼接@timestamp、message、instance_id三个字段，通过哈希算法生成ID，再把这个ID直接写入Bulk请求的index元数据中。

示例构造好的Bulk请求 #

POST /_bulk
{"index":{"_index":"eocs-technical-2022.08.24","_type":"_doc", "_id": "7a9f4d2b8e6c1a3e5f7b9d4c2a6e8f1d3b5a7c9e2d4f6a8b0c2e4d6f8a0b2d4"}}
{"@timestamp":"2022-08-24T13:49:34.428+0200","message":"This is testing message","hostname":"testcomputer.local","ip":"-","service_name":"test-service","instance_id":"c0","build.version":"master-d723731300570fd1b2d241c4849b223673d1c8d8","source":"com.example.ELKTest","level":"DEBUG","thread_name":"scheduler-1"}

代码示例（Python） #

用SHA256算法生成哈希ID：

import hashlib

def generate_custom_id(timestamp, message, instance_id):
    # 拼接字段并编码为字节
    combined_content = f"{timestamp}{message}{instance_id}".encode("utf-8")
    # 生成SHA256哈希并转为十六进制字符串
    return hashlib.sha256(combined_content).hexdigest()

# 针对单条文档生成ID
doc = {
    "@timestamp": "2022-08-24T13:49:34.428+0200",
    "message": "This is testing message",
    "instance_id": "c0",
    # 其他字段...
}
custom_id = generate_custom_id(doc["@timestamp"], doc["message"], doc["instance_id"])

# 构造Bulk条目
bulk_index_line = f'{{"index":{{"_index":"eocs-technical-2022.08.24","_type":"_doc", "_id": "{custom_id}"}}}}'
bulk_doc_line = str(doc).replace("'", '"')
bulk_request = f"{bulk_index_line}\n{bulk_doc_line}"

2. 使用Ingest Pipeline自动生成ID #

利用Elasticsearch的Ingest Pipeline功能，创建一个包含fingerprint处理器的管道，让Elasticsearch在文档写入前自动组合指定字段生成哈希ID，无需客户端处理。

步骤1：创建Ingest Pipeline #

PUT /_ingest/pipeline/generate-custom-id
{
  "processors": [
    {
      "fingerprint": {
        "fields": ["@timestamp", "message", "instance_id"],
        "target_field": "_id",
        "method": "sha256"
      }
    }
  ]
}

• fields：指定要组合的字段
• target_field：设为_id表示将哈希值作为文档ID
• method：指定哈希算法，可选md5、sha1、sha256等

步骤2：发送Bulk请求时指定管道 #

发送Bulk请求时，通过URL参数pipeline指定刚创建的管道，无需手动写_id：

POST /_bulk?pipeline=generate-custom-id
{"index":{"_index":"eocs-technical-2022.08.24","_type":"_doc"}}
{"@timestamp":"2022-08-24T13:49:34.428+0200","message":"This is testing message","hostname":"testcomputer.local","ip":"-","service_name":"test-service","instance_id":"c0","build.version":"master-d723731300570fd1b2d241c4849b223673d1c8d8","source":"com.example.ELKTest","level":"DEBUG","thread_name":"scheduler-1"}

可选：设置索引默认管道 #

如果该索引所有写入都需要用这个管道生成ID，可以将其设为索引的默认管道，后续发送Bulk请求无需再指定pipeline参数：

PUT /eocs-technical-2022.08.24/_settings
{
  "index.default_pipeline": "generate-custom-id"
}

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