除非应用一些过滤器，否则logstash会在事件到达时将其放入索引中。你的案子看起来很直截了当。如果您格式化sql查询以返回所需格式的数据，那么您不需要应用任何过滤器，只需连接数据库和sql查询以在logstash config中运行，并将输出作为ElasticSearch索引。
例如：
mysql查询看起来是这样的：（我不擅长mysql，下面只是给出一个想法-请验证它的工作原理）

SELECT team as id, 
       team, 
       GROUP_CONCAT(DISTINCT buyer SEPARATOR ', ') as buyers
FROM tablename GROUP BY team

这将返回如下结果：

+-----+------+------------------------+
| id  | team |         buyers         |
+-----+------+------------------------+
| one | one  | Q76876, Q66567, T99898 |
| two | two  | Q45456, S77676         |
+-----+------+------------------------+

logstash配置看起来很简单：

input {
  jdbc {
     jdbc_driver_library => "${DATABASE_DRIVER_PATH}"
     jdbc_driver_class => "${DATABASE_DRIVER_PATH}"
     jdbc_connection_string => "{CONNECTIONSTRING}"
     jdbc_user => "${DATABASE_USERNAME}"
     jdbc_password => "${DATABASE_PASSWORD}"
     statement_filepath => "${LOGSTASH_SQL_FILEPATH}" #this will be the sql written above
  }
}

filter {
}

output {
    elasticsearch {
        action => "index"       
        hosts => ["${ELASTICSEARCH_HOST}"]
        user => "${ELASTICSEARCH_USER}"
        password => "${ELASTICSEARCH_PASSWORD}"
        index => "${INDEX_NAME}"       
        document_type => "doc"                      
        document_id => "%{id}"       
    }
    stdout { codec => rubydebug }
    stdout { codec => dots }
}