我在MySQL中有一张表想要导入到Elasticsearch中

例如,数据看起来像这样

team   buyer
====   ======
one    Q76876
one    Q66567
one    T99898
two    Q45456
two    S77676

我想使用logstash将其导入elasticsearch并创建一个看起来像这样的索引
{
  "id": "one",
  "team": one,
  "buyers": ["Q76876", "Q66567", "T99898"]
},
{
  "id": "two",
  "team": "two",
  "buyers": ["Q45456", "S77676"]
}

我将如何编写.conf脚本来实现这一目标?

最佳答案

Logstash会在事件到达时将事件放入索引中,除非您应用一些过滤器。您的情况看起来很简单。如果您格式化sql查询以返回所需格式的数据,那么您无需应用任何过滤器,只需将数据库和sql查询挂接即可在logstash配置中运行,并将输出作为 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 }
}

09-10 16:26
查看更多