利用logstash從mysql同步數據到ElasticSearch

前面一篇已經把logstash和logstash-input-jdbc安裝好了。

下面就說下具體怎么配置。

1.先在安裝目錄bin下面（一般都是在bin下面）新建兩個文件jdbc.conf和jdbc.sql

2.配置jdbc.conf

input {
      stdin {
       }
      jdbc {
        # 連接的數據庫地址和哪一個數據庫，指定編碼格式，禁用SSL協議，設定自動重連
        jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/microstorage_backend?characterEncoding=UTF-8&useSSL=false&autoReconnect=true"
        jdbc_user => "root"
        jdbc_password => "123456"
        # 下載連接數據庫的驅動包，建議使用絕對地址
       jdbc_driver_library => "/usr/local/Cellar/logstash/6.5.4/libexec/logstash-core/lib/jars/mysql-connector-java-5.1.42.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   //這里如果是用時間追蹤比如：數據的更新時間或創建時間等和時間有關的這里一定不能是true, 切記切記切記，我是用update_time來追蹤的
        #追蹤的字段
     tracking_column => update_time
     record_last_run => true
     #上一個sql_last_value值的存放文件路徑, 必須要在文件中指定字段的初始值  這里說是必須指定初始值，我沒指定默認是1970-01-01 08：00：00 
     last_run_metadata_path => "/usr/local/opt/logstash/lastrun/.logstash_jdbc_last_run"  //這里的lastrun文件夾和.logstash_jdbc_last_run是自己創建的
 
      jdbc_default_timezone => "Asia/Shanghai"   //設置時區
      #statement => SELECT * FROM goods  WHERE update_time > :last_sql_value  //這里要說明一下如果直接寫sql語句，前面這種寫法肯定不對的
27 　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　，加上引號也試過也不對，所以我直接寫在jdbc.sql文件中
      statement_filepath => "/usr/local/Cellar/logstash/6.5.4/bin/jdbc.sql"
 
 
     #是否清除 last_run_metadata_path 的記錄,如果為真那么每次都相當於從頭開始查詢所有的數據庫記錄
     clean_run => false
 
       # 這是控制定時的，重復執行導入任務的時間間隔，第一位是分鍾 不設置就是1分鍾執行一次 35        schedule => "* * * * *"
       type => "std"
     }
 }

 filter {

    json {

        source => "message"

        remove_field => ["message"]

    }

}

output {

    elasticsearch {

        # 要導入到的Elasticsearch所在的主機

        hosts => "127.0.0.1:9200"

        # 要導入到的Elasticsearch的索引的名稱

        index => "goods"

        # 類型名稱（類似數據庫表名）

        document_type => "spu"

        # 主鍵名稱（類似數據庫主鍵）

        document_id => "%{id}"
    }

    stdout {

        # JSON格式輸出

        codec => json_lines

    }
}

3.配置jdbc.sql

select id,goods_name,goods_no,price,account_id,create_time,update_time from goods where update_time > :sql_last_value

4.我們來看下 .logstash_jdbc_last_run文件中的內容（網上講述該配置的時候都沒講到里面具體的內容寫法，導致很多人很迷惑，其中我就是）

前面的---具體什么意思，我也不太清楚。

 

5.啟動jdbc.conf配置，開始同步數據

 

第一次：因為時間是從1970年開始的所以會全部同步一遍。相當於全量同步了

從第二次開始，會從上次最新的一次時間同步，既新增和修改都會同步

 

遇到的問題：

1.ES中8小時時差的問題？

解決方法：從源頭解決問題

在jdbc.conf配置文件中只要是有關時間的字段都手動+8小時

 39　filter {
 40     json {
 41         source => "message"
 42         remove_field => ["message"]
 43     }
 44 　　 // date類型不能省略，不然會報錯，       就是把當前字段+8小時后賦值給新的字段，然后再取新字段的值賦值給老的字段，再把新的字段刪除
 45     date {
 46       match => ["message","UNIX_MS"]
 47       target => "@timestamp"
 48        }
 49          ruby {
 50                 code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
 51          }      
 52         ruby{   
 53                 code => "event.set('@timestamp',event.get('timestamp'))"
 54         }       
 55         mutate{ 
 56                remove_field => ["timestamp"]
 57         }      
 58         
 59    date {
 60     match => ["message","UNIX_MS"]
 61     target => "create_time"
 62          } 
 63          ruby {
 64                  code => "event.set('@create_time', event.get('create_time').time.localtime + 8*60*60)"
 65          }       
 66         ruby {   
 67                  code => "event.set('create_time',event.get('@create_time'))"
 68          }       
 69         mutate { 
 70          remove_field => ["@create_time"]
 71         }
 72         
 73         date {
 74         match => ["message","UNIX_MS"]
 75         target => "update_time"
 76          }
 77          ruby {
 78                  code => "event.set('@update_time', event.get('update_time').time.localtime + 8*60*60)"
 79          }       
 80         ruby {   
 81                  code => "event.set('update_time',event.get('@update_time'))"
 82          }       
 83         mutate {
 84          remove_field => ["@update_time"]
 85         }
 86   }

總結:主要是配置，有什么問題，先檢查配置文件。