使用logstash将数据库中的多张表数据同步到elasticsearch中

r囧r小猫 2022-03-07 15:38 414阅读 0赞

### 直接贴上配置文件,多张表的同步和单张表的同步只有配置文件的不同 ###

input {
        jdbc {
          # mysql 数据库链接,shop为数据库名
          jdbc_connection_string => "jdbc:mysql://数据库IP地址:3306/test?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC"
          # 用户名和密码
          jdbc_user => "root"
          jdbc_password => "数据库密码"
          # 驱动
          jdbc_driver_library => "/hw/elasticsearch/logstash-6.2.4/mysql/mysql-connector-java-5.1.44-bin.jar"
          # 驱动类名
          jdbc_driver_class => "com.mysql.jdbc.Driver"
          #是否分页
          jdbc_paging_enabled => "true"
          jdbc_page_size => "50000"
          #直接执行sql语句
          #statement =>"select * from test1;"
    	  # 执行的sql 文件路径+名称
          statement_filepath => "/hw/elasticsearch/logstash-6.2.4/bin/test.sql"
          #设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
          schedule => "* * * * *"
          # 索引类型
          #type => "jdbc"
        }
    jdbc {
          # mysql 数据库链接,shop为数据库名
          jdbc_connection_string => "jdbc:mysql://数据库IP地址:3306/test?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC"
          # 用户名和密码
          jdbc_user => "root"
          jdbc_password => "数据库密码"
          # 驱动
          jdbc_driver_library => "/hw/elasticsearch/logstash-6.2.4/mysql/mysql-connector-java-5.1.44-bin.jar"
          # 驱动类名
          jdbc_driver_class => "com.mysql.jdbc.Driver"
          #是否分页
          jdbc_paging_enabled => "true"
          jdbc_page_size => "50000"
          # 执行的sql 文件路径+名称
          statement =>"select * from test2"
          #statement_filepath => "/hw/elasticsearch/logstash-6.2.4/bin/test.sql"
          #设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
          schedule => "* * * * *"
          # 索引类型
          #type => "jdbc"
        }
    jdbc {
          # mysql 数据库链接,shop为数据库名
          jdbc_connection_string => "jdbc:mysql://数据库IP地址:3306/test?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC"
          # 用户名和密码
          jdbc_user => "root"
          jdbc_password => "数据库密码"
          # 驱动
          jdbc_driver_library => "/hw/elasticsearch/logstash-6.2.4/mysql/mysql-connector-java-5.1.44-bin.jar"
          # 驱动类名
          jdbc_driver_class => "com.mysql.jdbc.Driver"
          #是否分页
          jdbc_paging_enabled => "true"
          jdbc_page_size => "50000"
          # 执行的sql 文件路径+名称
          statement =>"select * from test3"
          #statement_filepath => "/hw/elasticsearch/logstash-6.2.4/bin/test.sql"
          #设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
          schedule => "* * * * *"
          # 索引类型
          #type => "jdbc"
        }
    
    }
     
    filter {
        json {
            source => "message"
            remove_field => ["message"]
        }
    #ruby { 
    #	code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)" 
    #	}
    #
    #	ruby{
    #	code => "event.set('day', (event.get('@timestamp').time.localtime + 8*60*60).strftime('%Y.%m.%d'))"
    #	}
    }
     
    output {
        elasticsearch {
    	#es的ip和端口
            hosts => ["http://192.168.217.155:9200"]
           	#ES索引名称（自己定义的）
    	index => "blog"
    	#文档类型
            document_type => "log"
    	#设置数据的id为数据库中的字段
            document_id => "%{id}"
        }
        stdout {
            codec => json_lines
        }
    }

### 只需要设置多个jdbc的模块就行了,有此需求的人直接将这个文件copy然后改改就能用 ###

### 点赞或者评论是我最大的动力,有问题欢迎留言或者联系q:1559810637   ###