Logstash实现PG数据同步到ES

1. Postgresql已经部署

2. Centos7.4以上

3. Docker及Docker-compose已经部署

1. 部署ELK，以下只展示ELK对应的compose部署yml文件

version: '3'
services:
  elasticsearch:
    image: elasticsearch:7.6.2
    container_name: elasticsearch
    environment:
      - "cluster.name=elasticsearch" #设置集群名称为elasticsearch
      - "discovery.type=single-node" #以单一节点模式启动
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m" #设置使用jvm内存大小
    volumes:
      - /data/elasticsearch/plugins:/usr/share/elasticsearch/plugins  # 插件文件挂载
      - /data/elasticsearch/data:/usr/share/elasticsearch/data  # 数据文件挂载
      - /data/elasticsearch/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml # 配置文件挂载
      - /etc/localtime:/etc/localtime
    ports:
      - 9200:9200
      - 9300:9300
  kibana:
    image: kibana:7.6.2
    container_name: kibana
    links:
      - elasticsearch:es #可以用es这个域名访问elasticsearch服务
    depends_on:
      - elasticsearch #kibana在elasticsearch启动之后再启动
    environment:
      - "elasticsearch.hosts=http://es:9200" #设置访问elasticsearch的地址
    ports:
      - 5601:5601
  logstash:
    image: logstash:7.6.2
    container_name: logstash
    volumes:
      - /data/logstash/logstash-springboot.conf:/usr/share/logstash/pipeline/logstash.conf #挂载logstash的配置文件
      - /data/logstash/drive:/home # 挂在一些用的jar和sql
      - /etc/localtime:/etc/localtime
    depends_on:
      - elasticsearch #kibana在elasticsearch启动之后再启动
    links:
      - elasticsearch:es #可以用es这个域名访问elasticsearch服务
    ports:
      - 4560:4560

2. 配置logstash-springboot.conf文件

input {
  tcp {
    mode => "server"
    host => "0.0.0.0"
    port => 4560
    codec => json_lines
	type => "log"
  }
  
  jdbc {
      # pg 数据库链接
      jdbc_connection_string => "jdbc:postgres://xx.xxx.xxx.xx:5432/test"
      # 用户名和密码
      jdbc_user => "postgres"
      jdbc_password => "123456"
      # 驱动
      jdbc_driver_library => "/home/postgresql-42.2.14.jar"
      # 驱动类名
      jdbc_driver_class => "org.postgresql.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
	  # 执行的sql 文件路径+名称
      statement_filepath => "/home/jdbc.sql"
      # 设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
	  schedule => "* * * * *"
      # 索引类型
	  type => "jdbc"
    }
}
output {

  if[type]=="log" {
    elasticsearch {
      hosts => "es:9200"
      index => "springboot-logstash-%{+YYYY.MM.dd}"
    }
  }
  
  if[type]=="jdbc" {
    elasticsearch {
      hosts => "es:9200"
      index => "test"
    }
  }
}`

3. 下载指定的依赖postgresql-42.2.14.jar,放到/home下

4. 编辑jdbc.sql（未加入增量）

select * from user_info