使用Logstash將MySQL資料同步到elasticsearch：

input {
    stdin {
    }
    jdbc {
        jdbc_driver_library => "mysql-connector-java-5.1.46-bin.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"

        #B表示你的MySQL的ip+3306，A表示某個資料庫
        jdbc_connection_string => "jdbc:mysql://B/A"

        #MySQL使用者名稱
 

        jdbc_user => "****"

        #密碼
        jdbc_password => "*****"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "100000"

        #這個引數表示你在開啟Logstash同步資料時需不需要clean掉上次的記錄
        clean_run => true

        #use_column_value 設定為true，表示按照增量更新模式更新資料
        use_column_value => true
 


        #這個引數表示將id作為每次增量更新所依照的引數
        tracking_column => "id"

        #這是存放上一次執行之後id的值
        last_run_metadata_path => "/usr/YEE/logstash-6.3.2/MySqlYee/****/*****.txt"

        #sql語句可以按照指令碼（路徑）或者sql語句給出
        #statement_filepath => "/usr/YEE/logstash-6.3.2/MySqlYee/****.sql"

        #sql_last_value每次讀取last_run_metadata_path中存放的值，下面語句增量更新是按照id值遞增的順序同步mysql中的內容
 

        statement => "select * from story where id > :sql_last_value"

        該引數按照 分 時 日 月 年 來設定多久更新一次資料，不設定表示僅更新一次；全* 表示每分鐘更新一次
        #schedule => "* * * * *"
   }
}
filter {
        json {
           source => "message"
           remove_field => ["message"]
        }
        #下面：當使用Logstash自動生成的mapping模板時過濾掉@timestamp和@version欄位
         mutate {
                remove_field =>
                ["@timestamp","@version"]
                }
   output {
        elasticsearch {
                #hosts：一般是localhost：9200
                hosts => ["****:9200"]
                index => "*****"

                #表示按照id同步mysql資料
                document_id => "%{id}"
                document_type => "****"

                #下面兩個引數表明載入我自己配置的Mapping模板，包括可以自行設定中文分詞等
                template_overwrite => true
                template => "/usr/YEE/logstash-6.3.2/MySqlYee/*****/template/*****_test1_ik.json"
        }

   stdout {
        codec => json_lines
   }
}

同步資料時可能遇到的一些問題：

• 對於全量更新，不用考慮資料更新的問題，但是資料量大的話需很長時間（我所描述的是每次都從elasticsearch刪除資料從零的更新，否則就會有資料重複的問題了）
• 增量更新對於某條已經存在資料內容修改的問題：
  
  • 通過在MySQL資料庫中新增一列用於記錄實時修改時間，然後上面增量更新的引數就設定為這個updatetime，每次資料有一點改動都要更新這個updatime的值，這樣就可以把每一次修改的值都更新；
• 增量更新對於資料刪除的問題：
  
  • 設定兩個引數：updatetime記錄當前更新資料時間；Delete_flag：表示這條資料有沒有被刪掉
• 當使用自己的template時遇到問題：elasticsearch每新增加一個index都會使用這個template中提供的mapping。Logstash同步資料時包括建立新的mapping時，都先檢視已經已經存在的template：GET /_cat/templates，然後就會看到一條記錄如下：
  
  通過程式碼：DELETE /_template/logstash刪掉這個template，然後在新增新的index
• Logstash could not be started because there is already another instance using the configured data directory.
  
  • 解決辦法：
  • 1、新增新的路徑
  • 2、 ps aux|grep logstash 查詢執行中的Logstash程序
    
    • kill -9 xxx kill程序