ElasticSearch（1）---Logstash同步Mysql數據到ElasticSearch

ElasticSearch同步Mysql

先講項目需求：對於資訊模塊添加搜索功能

這個搜索功能我就是采用ElasticSearch實現的，功能剛實現完，所以寫這篇博客做個記錄，讓自己在記錄下整個步驟和過程中的一些注意事項。

 

一、安裝elasticsearch和可視化工具

有關整個教程參考：mac安裝elasticsearch和可視化工具

1、安裝elasticsearch

網址地址：官網

2、安裝elasticsearch-head（可視化界面）

安裝地址：https://github.com/mobz/elasticsearch-head

3、安裝Node.js

安裝地址：Mac 下安裝node.js

4、grunt-cli（3、4主要配合2實現可視化界面）

命令：sudo npm install -g grunt-cli （我的是安裝在Mac上，所以其它不一定適用其它）

在終端運行: grunt --version（成功QQ圖）

5、elasticsearch和elasticsearch-head整合

修改 elasticsearch.yml 文件，在文檔的最末端加入

http.cors.enabled: true
http.cors.allow-origin: "*"

查看查看結果：輸入：localhost:9100

這里說明整個已經安裝成功並已經連接成功，green代表很健康

 

二、安裝logstash並同步MySQL數據庫

 相關博客推薦：安裝logstash並同步MySQL數據庫

1、下載logstash

注意：下載的版本要和你的elasticsearch的版本號一致，我的版本elasticsearch6.3.2

2、配置logstash-jdbc-input

據說2.x以上就不用配置了，不過我還是配置了

3、添加mysql-connector驅動jar包

把這個jar包放入logstash中：mysql-connector-java-5.1.21.jar

4、添加配置文件（用於連接elasticsearch和mysql數據庫）很重要！

具體的解釋推薦博客：logstash input jdbc連接數據庫

input {
  stdin {
  }

  jdbc {
  type => "news_info"
  #后面的test對應mysql中的test數據庫
  jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/news"
  jdbc_user => "root"
  jdbc_password => "root"
  tracking_column => "auto_id"
  record_last_run => "true"
  use_column_value => "true"
  #代表最后一次數據記錄id的值存放的位置，它會自動在bin目錄創建news,這個必填不然啟動報錯
  last_run_metadata_path => "news"
  clean_run => "false"

  # 這里代表mysql-connector-java-5.1.21.jar放在bin目錄
  jdbc_driver_library => "mysql-connector-java-5.1.21.jar"
  # the name of the driver class for mysql
  jdbc_driver_class => "Java::com.mysql.jdbc.Driver"
  jdbc_paging_enabled => "true"
  jdbc_page_size => "500"
  statement => "select auto_id,title,content,up_count,down_count,origin_create_time,grade from t_live_news_origin where auto_id > :sql_last_value and similar_score>0.5"
#定時字段 各字段含義（由左至右）分、時、天、月、年，全部為*默認含義為每分鍾都更新
  schedule => "* * * * *"
#設定ES索引類型
  }

    jdbc {
  type => "press_info"
  # mysql jdbc connection string to our backup databse 后面的test對應mysql中的test數據庫
  jdbc_connection_string => "jdbc:mysql:////127.0.0.1:3306/news"
  jdbc_user => "root"
  jdbc_password => "root"
  tracking_column => "auto_id"
  record_last_run => "true"
  use_column_value => "true"
  last_run_metadata_path => "news"
  clean_run => "false"
  jdbc_driver_library => "mysql-connector-java-5.1.21.jar"
  jdbc_driver_class => "Java::com.mysql.jdbc.Driver"
  jdbc_paging_enabled => "true"
  jdbc_page_size => "500"
  statement => "select auto_id,title,source_mc,read_count,summary,summary_img,origin_create_time from t_live_press_origin where auto_id > :sql_last_value"
#定時字段 各字段含義（由左至右）分、時、天、月、年，全部為*默認含義為每分鍾都更新
  schedule => "* * * * *"
  }
}

filter {
mutate {
  convert => [ "publish_time", "string" ]
 }

date {
  timezone => "Europe/Berlin"
  match => ["publish_time" , "ISO8601", "yyyy-MM-dd HH:mm:ss"]
}
#date {
 # match => [ "publish_time", "yyyy-MM-dd HH:mm:ss,SSS" ]
  # remove_field => [ "publish_time" ]
  # }
json {
  source => "message"
  remove_field => ["message"]
  }
}

output {

if [type]=="news_info" {
  elasticsearch {
#ESIP地址與端口
  hosts => "127.0.0.1:9200"
#ES索引名稱（自己定義的）
  index => "wantu_news_info"
#自增ID編號
  document_id => "%{auto_id}"
  }
}

if [type]=="press_info" {
  elasticsearch {
#ESIP地址與端口
  hosts => "127.0.0.1:9200"
#ES索引名稱（自己定義的）
  index => "wantu_press_info"
#自增ID編號
  document_id => "%{auto_id}"
  }
}

}

5、啟動logstash

#我這里的mysql.yml放到了bin的上層目錄
./logstash -f ../mysql.yml

6、實際效果

連接成功，已經成功把MySQL數據庫表中的數據存儲到Elasticsearch中，並且logstash每一分鍾去數據庫讀取最新數據。

最后看下我的logstash文件存放位置

 

三、坑和注意事項的總結  

1、如下報錯說明沒有找到你的mysql-connectorjar包，很可能你的jar沒有放到配置文件指定的目錄。 

2、需要重新讓查詢從0開始。

那就刪除last_run_metadata_path => "news"的news文件，當然也要記得刪除該索引好讓它重新讀取數據庫表中數據。

相關其它坑博客地址：坑的總結

再遇到其它相關坑，百度吧，都能快速找到答案。

 

 我只是偶爾安靜下來，對過去的種種思忖一番。那些曾經的舊時光里即便有過天真愚鈍，也不值得譴責。畢竟，往后的日子，還很長。不斷鼓勵自己，

 天一亮，又是嶄新的起點，又是未知的征程（上校4）