Logstash同步多數據庫多表到ES的配置文件

input {
    stdin{
    }
    jdbc {
      # 連接的數據庫地址和哪一個數據庫，指定編碼格式，禁用SSL協議，設定自動重連
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/userdb?characterEncoding=UTF-8&useSSL=false&autoReconnect=true"
      # 用戶名密碼
      jdbc_user => "user"
      jdbc_password => "password"
      
      #以下配置是為了達到增量同步的目的，每次同步結束之后會記錄最后一條數據的tracking_column列，這里設置的是id，就會將這個值記錄在last_run_metadata_path中。下次在執行同步的時候會將這個值，賦給sql_last_value
      
      # 是否記錄上次執行結果，true表示會將上次執行結果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
      record_last_run => "true"
      # 需要記錄查詢結果某字段的值時，此字段為true，否則默認tracking_column為timestamp的值；
      use_column_value => "true"
      # 需要記錄的字段，用於增量同步，需是數據庫字段
      tracking_column => "id"
      # record_last_run上次數據存放位置；
      last_run_metadata_path => "/u02/tomcat-8.5.9/logstash-7.6.2/bin/user_last_id"
      # 是否清除last_run_metadata_path的記錄，需要增量同步時此字段必須為false；
      clean_run => "false"
      
      # jar包的位置
      jdbc_driver_library => "/u02/tomcat/mysql-connector-java-8.0.20/mysql-connector-java-8.0.20.jar"
      # mysql的Driver
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_default_timezone => "Asia/Shanghai"
      # 開啟分頁查詢（默認false不開啟）
      jdbc_paging_enabled => "true"
      # 單次分頁查詢條數（默認100000,若字段較多且更新頻率較高，建議調低此值）
      jdbc_page_size => "500"
      #注意這個sql不能出現type，這是es的保留字段
      statement => "select * from user where id > :sql_last_value ORDER BY  id asc  limit  20000"
      # 同步頻率(分 時 天 月 年)，默認每分鍾同步一次；當前配置每10分鍾執行一下 
      schedule => "*/10 * * * *"
    }
    jdbc {
      # 連接的數據庫地址和哪一個數據庫，指定編碼格式，禁用SSL協議，設定自動重連
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/userdb?characterEncoding=UTF-8&useSSL=false&autoReconnect=true"
      # 用戶名密碼
      jdbc_user => "user"
      jdbc_password => "password"
      
      
      #這些配置是為了達到增量同步的目的，每次同步結束之后會記錄最后一條數據的tracking_column列，這里設置的是id，就會將這個值記錄在last_run_metadata_path中。下次在執行同步的時候會將這個值，賦給sql_last_value
      
      # 是否記錄上次執行結果，true表示會將上次執行結果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
      record_last_run => "true"
      # 需要記錄查詢結果某字段的值時，此字段為true，否則默認tracking_column為timestamp的值；
      use_column_value => "true"
      # 需要記錄的字段，用於增量同步，需是數據庫字段
      tracking_column => "id"
      # record_last_run上次數據存放位置；
      last_run_metadata_path => "/u02/tomcat-8.5.9/logstash-7.6.2/bin/dept_last_id"
      # 是否清除last_run_metadata_path的記錄，需要增量同步時此字段必須為false；
      clean_run => "false"
      
      # jar包的位置
      jdbc_driver_library => "/u02/tomcat/mysql-connector-java-8.0.20/mysql-connector-java-8.0.20.jar"
      # mysql的Driver
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_default_timezone => "Asia/Shanghai"
      # 開啟分頁查詢（默認false不開啟）
      jdbc_paging_enabled => "true"
      # 單次分頁查詢條數（默認100000,若字段較多且更新頻率較高，建議調低此值）
      jdbc_page_size => "500"
      #注意這個sql不能出現type，這是es的保留字段
      statement => "select * from dept where id > :sql_last_value   ORDER BY  id asc  limit  20000"
      schedule => "*/10 * * * *"
    }
}

output {
    if [type]=="user" {
        elasticsearch {
            hosts => "127.0.0.1:9200"
            # index名
            index => "monitor_user"
            # type名
            document_type => "_doc"
            # 數據唯一索引（建議使用數據庫KeyID）
            document_id => "%{id}"
        }
        # JSON格式輸出
        stdout {
            codec => json_lines
        }
    }

    if [type]=="dept" {
        elasticsearch {
            hosts => "127.0.0.1:9200"
            # index名
            index => "monitor_dept"
            # type名
            document_type => "_doc"
            # 數據唯一索引（建議使用數據庫KeyID）
            document_id => "%{id}"
        }
        # JSON格式輸出
        stdout {
            codec => json_lines
        }
    }
    
}