logstash同步pgsql数据到Elasticsearch

一、对于logstash的配置我就不在多说，主要是三部分，input、filter、output的配置

二、配置步骤

1、input配置

input { stdin { } jdbc { jdbc_connection_string => "jdbc:postgresql://127.0.0.1:5432/world" jdbc_user => "postgres" jdbc_password => "zhang123" jdbc_driver_library => "D:\logstash-6.4.0\bin\pgsql\postgresql-42.2.5.jar" jdbc_driver_class => "org.postgresql.Driver" jdbc_paging_enabled => "true" jdbc_page_size => "300000" use_column_value => "true" tracking_column => "id" statement_filepath => "D:\logstash-6.4.0\bin\pgsql\jdbc.sql" schedule => "* * * * *" type => "jdbc" jdbc_default_timezone =>"Asia/Shanghai" } }

2、filter配置

filter { json { source => "message" remove_field => ["message"] } }

3、output 配置，就是elasticsearch的基本配置

output { elasticsearch { hosts => ["localhost:9200"] index => "test_out" template => "D:\logstash-6.4.0\bin\pgsql\es-template.json" template_name => "t-statistic-out-logstash" template_overwrite => true document_type => "out" document_id => "%{id}" } stdout { codec => json_lines } }

以上就是整个logstash 的jdbc.conf

4、es-template.json的配置

{ "template" : "t-statistis-out-template", "order":1, "settings": { "index": { "refresh_interval": "5s" } }, "mappings": { "_default_": { "_all" : {"enabled":false}, "dynamic_templates": [ { "message_field" : { "match" : "message", "match_mapping_type" : "string", "mapping" : { "type" : "string", "index" : "not_analyzed" } } }, { "string_fields" : { "match" : "*", "match_mapping_type" : "string", "mapping" : { "type" : "string", "index" : "not_analyzed" } } } ], "properties": { "@timestamp": { "type": "date" }, "@version": { "type": "keyword" }, "id": { "type": "keyword" }, "name": { "type": "keyword" }, "pp": { "type": "keyword" } } } }, "aliases": {} }

最后就是就是下载好pgsql的连接驱动，这个官网可以下载；配置好自己的数据库表格的数据

启动命令：进入到logstash的bin目录下，自己的logstash配置都是放在bin的pgsql这个目录下面（这个自己随意创建位置都可以）

logstash.bat -f ./pgsql/jdbc.conf