时间戳

首页 > 解决方案 > 使用 logstash 在弹性搜索中将两个索引合并为第三个索引

elasticsearch - 使用 logstash 在弹性搜索中将两个索引合并为第三个索引

问题描述

我有两个索引

  1. 员工数据 {"code":1, "name":xyz, "city":"Mumbai" }
  2. 交易数据 {"code":1, "Month":June", payment:78000 }

我想要这样的第三个索引 3)join_index

{"code":1, "name":xyz, "city":"Mumbai", "Month":June", payment:78000 } 怎么可能？？

我正在尝试使用logstash

input {
  elasticsearch {
    hosts => "localost"
    index => "employees_data,transaction_data"
   
     query => '{ "query": { "match": { "code": 1} } }'
    scroll => "5m"
    docinfo => true
  }
}
output {

弹性搜索 { 主机 => [“本地主机”]

index => "join1"
   }

}

标签： elasticsearchindexinglogstashelasticsearch-5

解决方案

您不需要 Logstash 来执行此操作，Elasticsearch 本身通过利用enrich processor.

首先，您需要创建一个丰富的策略（使用最小的索引，假设它是employees_data ）：

PUT /_enrich/policy/employee-policy
{
  "match": {
    "indices": "employees_data",
    "match_field": "code",
    "enrich_fields": ["name", "city"]
  }
}

然后您可以执行该策略以创建丰富索引

POST /_enrich/policy/employee-policy/_execute

创建并填充丰富索引后，下一步需要您创建一个使用上述丰富策略/索引的摄取管道：

PUT /_ingest/pipeline/employee_lookup
{
  "description" : "Enriching transactions with employee data",
  "processors" : [
    {
      "enrich" : {
        "policy_name": "employee-policy",
        "field" : "code",
        "target_field": "tmp",
        "max_matches": "1"
      }
    },
    {
      "script": {
        "if": "ctx.tmp != null",
        "source": "ctx.putAll(ctx.tmp); ctx.remove('tmp');"
      }
    }
  ]
}

最后，您现在已准备好使用连接数据创建目标索引。只需将_reindexAPI 与我们刚刚创建的摄取管道结合使用即可：

POST _reindex
{
  "source": {
    "index": "transaction_data"
  },
  "dest": {
    "index": "join1",
    "pipeline": "employee_lookup"
  }
}

运行后，join1索引将包含您需要的内容，例如：

  {
    "_index" : "join1",
    "_type" : "_doc",
    "_id" : "0uA8dXMBU9tMsBeoajlw",
    "_score" : 1.0,
    "_source" : {
      "code":1, 
      "name": "xyz", 
      "city": "Mumbai", 
      "Month": "June", 
      "payment": 78000 
    }
  }

推荐阅读