應對sharding-jdbc結合mybatis實現分庫分表功能  分表的聯合查詢採用將mysql的資料同步到elasticsearch進行篩選

安裝操作指南：（1）、（2）

其中windows目錄展示如下：

版本控制：1. 需要jdk:1.8（1.8.0_60）

                  2.ElasticSearch(5.6.3)

                  3.建議logstash版本與ElasticSearch版本號對應一致（5.6.3）

                  4.ruby-2.4.4-2-x64

說明：

ElasticSearch配置head外掛：

1.head在es2.X及以前是作為es的一個外掛存在的,可以直接通過plugin命令安裝,5.x及以後需要作為一個單獨的服務執行,基於nodejs環境及grunt啟動服務

2.logstash + Logstash-input-jdbc來同步資料,logstash-input-jdbc外掛是logstash 的一個個外掛,使用ruby語言開發。所以要先安裝ruby,也是為了好使用ruby中的gem安裝外掛，下載地址: https://rubyinstaller.org/downloads/

核心：同步方式

1、資料同步方式

      全量同步與增量同步

     全量同步是指全部將資料同步到es，通常是剛建立es，第一次同步時使用。增量同步是指將後續的更新、插入記錄同步到es。

2. logstash-input-jdbc使用

官方文件地址 

logstash目錄下配置檔案的結構：

全量同步配置檔案：jdbc.conf   + jdbc.sql + mysql-connector-java-5.1.9

增量同步配置檔案：jdbc_add.conf   + jdbc_add.sql + mysql-connector-java-5.1.9 + station_parameter.txt

jdbc..conf：

input {
    stdin {
    }
    jdbc {
      # mysql 資料庫連結,test為資料庫名
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/test"
      # 使用者名稱和密碼
      jdbc_user => "root"
      jdbc_password => "root"
      # 驅動
      jdbc_driver_library => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\mysql-connector-java-5.1.9.jar"
      # 驅動類名
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
	  # 執行的sql 檔案路徑+名稱
      statement_filepath => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\jdbc.sql"
      # 設定監聽間隔  各欄位含義（由左至右）分、時、天、月、年，全部為*預設含義為每分鐘都更新
	  schedule => "* * * * *"
      # 索引型別
	  type => "jdbc"
    }
}


filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}


output {
    elasticsearch {
	    # ES的IP地址及埠
        hosts => ["localhost:9200"]
	    # 索引名稱
        index => "article"
	    # 自增ID 需要關聯的資料庫中有有一個id欄位，對應索引的id號
        document_id => "%{id}"
    }
    stdout {
	   # JSON格式輸出
        codec => json_lines
    }
}
 

 jdbc.sql：

select * from t_student_01

增量同步jdbc_add.sql：

input {
    stdin {
    }
    jdbc {
      # mysql 資料庫連結,test為資料庫名
      jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/test"
      # 使用者名稱和密碼
      jdbc_user => "root"
      jdbc_password => "root"
      # 驅動
      jdbc_driver_library => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\mysql-connector-java-5.1.9.jar"
      # 驅動類名
      jdbc_driver_class => "com.mysql.jdbc.Driver"

      #處理中文亂碼問題
      codec => plain { charset => "UTF-8"}
      #使用其它欄位追蹤，而不是用時間
      use_column_value => true
      #追蹤的欄位
      tracking_column => id
      record_last_run => true
     #上一個sql_last_value值的存放檔案路徑, 必須要在檔案中指定欄位的初始值
     last_run_metadata_path => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\station_parameter.txt"
     #開啟分頁查詢
     jdbc_paging_enabled => true
     jdbc_page_size => 300
	  
      # 執行的sql 檔案路徑+名稱
      statement_filepath => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\jdbc.sql"
      # 設定監聽間隔  各欄位含義（由左至右）分、時、天、月、年，全部為*預設含義為每分鐘都更新
	  schedule => "* * * * *"
      # 索引型別
	  type => "jdbc"

    }
}

filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}

output {
    elasticsearch {
	    # ES的IP地址及埠
        hosts => ["localhost:9200"]
		# 索引名稱
        index => "article"
		# 自增ID
        document_id => "%{id}"
    }
    stdout {
	    # JSON格式輸出
        codec => json_lines
    }
}
引數介紹:

//是否記錄上次執行結果, 如果為真,將會把上次執行到的 tracking_column 欄位的值記錄下來,儲存到 last_run_metadata_path 指定的檔案中
record_last_run => true

//是否需要記錄某個column 的值,如果 record_last_run 為真,可以自定義我們需要 track 的 column 名稱，此時該引數就要為 true. 否則預設 track 的是 timestamp 的值.
use_column_value => true

//如果 use_column_value 為真,需配置此引數. track 的資料庫 column 名,該 column 必須是遞增的.比如：ID.
tracking_column => MY_ID

//指定檔案,來記錄上次執行到的 tracking_column 欄位的值
//比如上次資料庫有 10000 條記錄,查詢完後該檔案中就會有數字 10000 這樣的記錄,下次執行 SQL 查詢可以從 10001 條處開始.
//我們只需要在 SQL 語句中 WHERE MY_ID > :sql_last_value 即可. 其中 :sql_last_value 取得就是該檔案中的值(10000).(！！！注意這裡唯一不變就是 sql_last_value 是資料庫表的主鍵的當前最大值)
last_run_metadata_path => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\station_parameter.txt"


//是否清除 last_run_metadata_path 的記錄,如果為真那麼每次都相當於從頭開始查詢所有的資料庫記錄
clean_run => false

//是否將 column 名稱轉小寫
lowercase_column_names => false

//存放需要執行的 SQL 語句的檔案位置
statement_filepath => "G:\Developer\Elasticsearch5.5.1\ES5\logstash-5.5.1\bin\mysql\jdbc.sql"

增量同步：jdbc_add.sql:

select * from t_student_01 where id > :sql_last_value

增量同步：station_parameter.txt（！注意  #上一個sql_last_value值的存放檔案路徑, 起初必須要在檔案中指定欄位的初始值，其中“---”這符號是自己生成的）

最終效果： 

這個檔案裡記錄上次執行到的 tracking_column 欄位的值,比如上次資料庫有 10000 條記錄,查詢完後該檔案中就會有數字 10000 這樣的記錄,下次執行 SQL 查詢可以從 10001 條處開始,我們只需要在 SQL 語句中 WHERE MY_ID > :sql_last_value 即可. 其中 :sql_last_value 取得就是該檔案中的值。

遇到的問題：

（1）在安裝好ElasticSearch的head外掛後總是用grunt server命令啟動時總是報錯：

解決方案： 命令：

npm install grunt --save-dev

（2）增量同步中上述jdbc_add.sql檔案中的 記錄tracking_column 欄位的值,  必須是 

where id > :sql_last_value

（3）上述所有配置檔案需要編碼是ANSI編碼方式（一般預設是UTF-8編碼），否則出錯