將mysql數據同步到ES6.4（全量+增量）

下載安裝包時注意下載到指定文件夾 這裏我放在OPT文件夾下
一：安裝logstash
進入到opt文件夾打開終端 執行如下命令
wget -c https://artifacts.elastic.co/downloads/logstash/logstash-6.4.0.zip
加上-c支持斷點續傳
二：解壓logstash
unzip logstash-6.4.0.zip
三：進入到logstash bin目錄
cd logstash-6.4.0/bin
四：安裝logstash-jdbc
./logstash-plugin install logstash-input-jdbc
五：編寫配置文件（jdbc.sql和jdbc.conf，建議在bin目錄下vim jdbc.conf)
六：首先在bin目錄下新建一個mysql目錄,裏面包含jdbc.conf,jdbc.sql文件,加入mysql的驅動
jdbc.conf內容：

input {
    stdin {
    }
    jdbc {
      # 鏈接的數據庫地址和哪個數據庫，指定編碼格式，禁用SSL協議，設定自動重連
      jdbc_connection_string => "jdbc:mysql://數據庫地址:端口號/數據庫名?characterEncoding=UTF-8&useSSL=false&autoReconnect=true"
      # 你的帳戶密碼
      jdbc_user => "帳號"
      jdbc_password => "密碼"
      # 鏈接數據庫的驅動包，建議使用絕對地址
      jdbc_driver_library => "mysql/mysql-connector-java-5.1.45-bin.jar"
      # 這是不用動就好
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"

    #處理中文亂碼問題
      codec => plain { charset => "UTF-8"}

       #使用其它字段追蹤，而不是用時間
      use_column_value => true
       #追蹤的字段      
    tracking_column => testid      
    record_last_run => true     
    #上一個sql_last_value值的存放文件路徑, 必需要在文件中指定字段的初始值     
    last_run_metadata_path => "mysql/station_parameter.txt"

      jdbc_default_timezone => "Asia/Shanghai"
 
      statement_filepath => "mysql/jdbc.sql"
      

    #是否清除 last_run_metadata_path 的記錄,若是爲真那麼每次都至關於從頭開始查詢全部的數據庫記錄
    clean_run => false

      # 這是控制定時的，重複執行導入任務的時間間隔，第一位是分鐘
      schedule => "* * * * *"
      type => "jdbc"
    }
}


filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}
 
 
output {
    elasticsearch {
        # 要導入到的Elasticsearch所在的主機
        hosts => "192.168.105.180:9200"
        # 要導入到的Elasticsearch的索引的名稱
        index => "db_anytest"
        # 類型名稱（相似數據庫表名）
        document_type => "table_anytest"
        # 主鍵名稱（相似數據庫主鍵）
        document_id => "%{testid}"
        
    }

    stdout {
        # JSON格式輸出
        codec => json_lines
    }
    
}

 

jdbc.sql裏面就直接寫sql語句就好了

全量同步sql語句就不要寫testid了 增量同步就加上where條件 where testid > :sql_last_value
七：開始導入(須要用root帳戶啓動)
./logstash -f mysql/jdbc.conf