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

如何解决使用logstash在弹性搜索中将两个索引合并为第三索引

我有两个索引

1. employee_data {"code":1,"name":xyz,"city":"Mumbai" }
2. transaction_data {"code":1,"Month":June",payment:78000 }

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

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

我正在尝试使用logstash

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

elasticsearch { 主机=> [“”本地主机“]

index => "join1"
   }

}

解决方法

您可以在 employees_data

上使用elasticsearch 输入

在您的过滤器中，对 transaction_data

使用elasticsearch 过滤器

input {
  elasticsearch {
    hosts => "localost"
    index => "employees_data"
   
     query => '{ "query": { "match_all": { } } }'
     sort => "code:desc"

    scroll => "5m"
    docinfo => true
  }
}
filter {
    elasticsearch {
              hosts => "localhost"
              index => "transaction_data"
              query => "(code:\"%{[code]}\"
              fields => { 
                    "Month" => "Month","payment" => "payment" 
                   }
        }
}
output {
  elasticsearch { 
    hosts => ["localhost"]
    index => "join1"
   }
}

然后使用elasticsearch 输出

将新文档发送到第三个索引

您将拥有3个弹性搜索连接，结果可能会有点慢。 但这有效。

,

您不需要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');"
      }
    }
  ]
}

最后，您现在可以使用连接的数据创建目标索引了。只需将_reindex API与我们刚刚创建的摄取管道结合使用：

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 
    }
  }

,

据我所知，仅使用Elasticsearch API不会发生这种情况。要处理此问题，您需要为相关文档设置唯一的ID。例如，您在问题中提到的代码可以是文档的良好ID。因此，您可以将第一个索引重新索引为第三个索引，并使用UPDATE API通过从第二个索引读取文档来更新它们，并通过其ID将它们更新为第三个索引。希望我能帮上忙。