Document_id

Добрый день!
Имеется индекс test, в нем несколько тегов: tags1, tags2, tags3.
Данные я получаю из jdbc запроса к БД.
В output я меняю _id elasticsearch => на document_id и отправляю это все в один индекс test. В результате в elasticsearch получаю рандомное количество документов по каждому tags отличное от того, что получаю в запросе к БД.

Если я делаю output меняя _id elasticsearch => на document_id, но отправляю это все в разные индексы test1, test2, test3 - то в результате количество документов по каждому тегу совпадает с запросами к БД.

Есть ли возможность сделать output в один индекс test, при этом изменив _id elasticsearch => на document_id?

>   jdbc {
    jdbc_driver_library => "/usr/share/logstash/sqljdbc_7.2/mssql-jdbc-7.2.2.jre8.jar"
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    jdbc_connection_string => "jdbc:sqlserver://1.2.3.4:1234;databaseName=My_DB"
    jdbc_user => "logstash"
    jdbc_password => "**********"
    schedule => "0 3 * * *"
    statement_filepath => "/opt/logstash/tags1.sql"
    tracking_column => "my_id"
    tracking_column_type => "timestamp"
    last_run_metadata_path => "/opt/logstash/tags1"
    use_column_value => true
    tags => ["tags1"]
  }
>   jdbc {
    jdbc_driver_library => "/usr/share/logstash/sqljdbc_7.2/mssql-jdbc-7.2.2.jre8.jar"
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    jdbc_connection_string => "jdbc:sqlserver://1.2.3.4:1234;databaseName=My_DB"
    jdbc_user => "logstash"
    jdbc_password => "**********"
    schedule => "2 3 * * *"
    statement_filepath => "/opt/logstash/tags2.sql"
    tracking_column => "my_id"
    tracking_column_type => "timestamp"
    last_run_metadata_path => "/opt/logstash/tags2"
    use_column_value => true
    tags => ["tags2"]
  }
....

> output 
{
  if ("tags1" in [tags]
    ) {
    elasticsearch {
      hosts => [ "localhost:9200" ]
      index => "test"
      user => user
      password => *********
      document_id => "%{my1_id}"
      doc_as_upsert => true
      action => "update"
    }
  }
  if ("tags2" in [tags]
    ) {
    elasticsearch {
      hosts => [ "localhost:9200" ]
      index => "test"
      user => user
      password => *********
      document_id => "%{my2_id}"
      doc_as_upsert => true
      action => "update"
    }
  }
  if ("tags3" in [tags]
    ) {
    elasticsearch {
      hosts => [ "localhost:9200" ]
      index => "test"
      user => user
      password => *********
      document_id => "%{my3_id}"
      doc_as_upsert => true
      action => "update"
    }
  }
}

У вас, скорее всего, id документов с разными тэгами пересекаются, и когда вы сваливаете все в один индекс одни документы переписывают уже существующие. Самый простой способ - добавить значение тэгов к id документа, чтобы у всех документов был уникальный id.

Спасибо за ответ! Данный способ не проверил, ограничился тем, что сделал несколько индексов test1, test2, test3 и объединил их в один шаблон test*.