1、我使用的是windows 版本的logstash-7.5.1 进入 logstash-7.5.1\bin，直接 点击logstash.bat运行即可即可。
2、在bin 文件夹中建立pgsql 文件夹（随意位置，随意名字）
3、首先要连接postgreSQL 数据库，需要pg引擎，我使用的是postgresql-42.2.8.jar
放到pgsql文件夹中
然后建立logstas-pgsql.conf

input {
    stdin {
    }
    jdbc {
      jdbc_connection_string => "jdbc:postgresql://127.0.0.1:5432/testdata"
      jdbc_user => "***"
      jdbc_password => "******" 更换自己的数据库用户名和密码
      jdbc_driver_library => "******\logstash-7.5.1\bin\pgsql\postgresql-42.2.8.jar" 更换自己的驱动地址
      jdbc_driver_class => "org.postgresql.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "300000"
      use_column_value => "true"
      tracking_column => "id"
      statement_filepath => "******\logstash-7.5.1\bin\pgsql\logstash-pgsql.sql" 这是要操作的sql 表，下面会提供
          schedule => "* * * * *"
          type => "jdbc"
          jdbc_default_timezone =>"Asia/Shanghai"
    }
}
filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}
output {
    elasticsearch {
        hosts => ["localhost:9200"]
        index => "test_out"
              template => "****\logstash-7.5.1\bin\pgsql\es-template.json"   es 索引模板，可不定义
              template_name => "t-statistic-out-logstash"
              template_overwrite => true
              document_type => "out"
        document_id => "%{id}"
    }
    stdout {
        codec => json_lines
    }
}

logstash-pgsql.sql 文件内容非常简单

select * from tb_table

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": {}
}

4、进入 bin 目录

运行命令

logstash -f ./pgsql/logstash-pgsql.conf

注意：启动时因为是同台机器运行多个logstash实例， 需要指定不同的数据存储目录 path.Data

执行命令：

logstash -f ./pgsql/logstash-pgsql.conf --path.data=/pgconfig/