全文搜索引擎 Elasticsearch （三）logstash-input-jdbc同步數據 到elasticsearch

參考鏈接：

1， 源碼地址，官方介紹

2， logstash-input-jdbc使用建議

3, 官網介紹例子，使用 logstash-input-jdbc 到 elasticsearch

一、安裝 elastisearch, logstash

參考我的上一篇鏈接 ：centos下 ELK部署文檔

二、安裝logstash-input-jdbc插件

logstash-input-jdbc插件是logstash 的一個個插件。

使用ruby語言開發。

下載插件過程中最大的坑是下載插件相關的依賴的時候下不動，因為國內網絡的原因，訪問不到亞馬遜的服務器。

解決辦法，改成國內的ruby倉庫鏡像。此鏡像托管於淘寶的阿里雲服務器上 ：

淘寶鏡像官網地址

1， 如果沒有安裝 gem 的話 安裝gem 

sudo yum install gem

替換淘寶

1，gem sources --add https://ruby.taobao.org/ --remove https://rubygems.org/
2，gem sources -l

*** CURRENT SOURCES ***

https://ruby.taobao.org
# 請確保只有 ruby.taobao.org
如果 還是顯示 https://rubygems.org/ 進入 home的 .gemrc 文件
sudo vim ~/.gemrc 
手動刪除 https://rubygems.org/

2， 修改Gemfile的數據源地址。步驟：

1， whereis logstash # 查看logstash安裝的位置， 我的在 /opt/logstash/ 目錄
2， sudo vi Gemfile # 
修改 source 的值 為： "https://ruby.taobao.org"
3, sudo vi Gemfile.jruby-1.9.lock # 找到 remote 修改它的值為： https://ruby.taobao.org

或者直接替換源這樣你不用改你的 Gemfile 的 source。

sudo gem install bundler


$ bundle config mirror.https://rubygems.org https://ruby.taobao.org

安裝logstash-input-jdbc 

我一共試了三種方法，一開始都沒有成功，原因如上，鏡像的問題。一旦鏡像配置成淘寶的了，理論上隨便選擇一種安裝都可以成功，我用的是第三種。

第一種：

cd /opt/logstash/

sudo bin/plugin install logstash-input-jdbc

如果成功就成功了。

第二種：

1，進入源碼地址的release頁面logstash-input-jdbc

選擇對應的版本。我的logstash版本是1.4.0，對應的插件版本是1.0.0

關於插件版本的選擇 參考這里：這是ruby Gemfile所有插件的官方地址，參考logstash-core ，如果你的logstash版本跟他對應上，那么這個插件的版本就是你要的版本。

2，復制 1.0.0的下載地址

cd /opt/logstash
sudo wget https://github.com/logstash-plugins/logstash-input-jdbc/archive/v1.0.0.zip
sudo bin/plugin install v1.0.0.zip

如果成功了就成功了

3， 遺憾的是上兩個步驟都沒成功，我是手動裝的。

參考 安裝jdbc插件的問題, 這篇文章的最后 ，這為好心的姑娘，也建議改變gemgile里的鏡像，我估計這姑娘是chinese。

$ cd /opt/logstash
$ wget https://github.com/logstash-plugins/logstash-input-jdbc/archive/v1.0.0.zip
$ unzip master.zip
$ cd logstash-input-jdbc-master

注意，有坑：
1， 修該 里面的 Gemfile 
sudo vi Gemfile  
修改 source 的值 為： "https://ruby.taobao.org"

2，修改 logstash-input-jdbc.gemspec
sudo vi logstash-input-jdbc.gemspec
找到 
    s.files = `git ls-files`.split($\)
改為：
    s.files = [".gitignore", "CHANGELOG.md", "Gemfile", "LICENSE", "NOTICE.TXT", "README.md", "Rakefile", "lib/logstash/inputs/jdbc.rb", "lib/logstash/plugin_mixins/jdbc.rb", "logstash-input-jdbc.gemspec", "spec/inputs/jdbc_spec.rb"]
OK! 繼續.....

$ gem build logstash-input-jdbc.gemspec
(output:
fatal: Not a git repository (or any of the parent directories): .git
Successfully built RubyGem
Name: logstash-input-jdbc
Version: 1.0.0
File: logstash-input-jdbc-1.0.0.gem)
$ mv logstash-input-jdbc-1.0.0.gem /opt/logstash/
$ cd ..
$ bin/plugin install logstash-input-jdbc-1.0.0.gem

如果再不成功 我也沒招了。

三、實現樣例。

假如上面步驟都搞定了...重點來了 繼續看...沒搞定也可以接着看啦..hahahaha....實戰......

目的 ： 監聽數據表的數據，當我有新增時增加到elasticsearch，當我修改時，update到elasticsearch。

第一 前提：

1, 我有mysql數據庫，我有一張hotel 表， hotel_account表（此表里有hotel_id）, 里面無數據。

2，已經啟動 elasticsearch . 地址是 http://192.168.0.45 端口：9200.

3，已經安裝 logstash, 地址在 /opt/logstash

第二 准備

兩個文件： jdbc.conf  

                   jdbc.sql 。名字隨便起啦。

  一個 mysql 的java 驅動包  ： mysql-connector-java-5.1.36-bin.jar

jdbc.conf 內容：

注意 statement_filepath => "jdbc.sql" 這個名字要跟下面的sql文件的名字對應上。

參考 logstash-input-jdbc官方參考文檔

重點字段說明：

schedule：設置監聽間隔。可以設置每隔多久監聽一次什么的。具體參考官方文檔。

statement_filepath： 執行的sql 文件路徑+名稱

 

input {
    stdin {
    }
    jdbc {
      # mysql jdbc connection string to our backup databse
      jdbc_connection_string => "jdbc:mysql://192.168.0.49:3306/dfb"
      # the user we wish to excute our statement as
      jdbc_user => "test"
      jdbc_password => "test"
      # the path to our downloaded jdbc driver
      jdbc_driver_library => "/opt/logstash/mysql-connector-java-5.1.36-bin.jar"
      # the name of the driver class for mysql
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement_filepath => "jdbc.sql"
      schedule => "* * * * *"
      type => "jdbc"
    }
}

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

output {
    elasticsearch {
        host => "192.168.0.199"
        port => "9200"
        protocol => "http"
        index => "mysql01"
        document_id => "%{id}"
        cluster => "logstash-elasticsearch"
    }
    stdout {
        codec => json_lines
    }
}

 

jdbc.sql 

select 
    h.id as id, 
    h.hotel_name as name, 
    h.photo_url as img,
    ha.id as haId, 
    ha.finance_person
from 
    hotel h LEFT JOIN hotel_account ha on h.id = ha.hotel_id
where 
    h.last_modify_time >= :sql_last_start

第三： 啟動 logstash

sudo bin/logstash -f jdbc.conf

如果一切順利 應該如圖：

現在 logstash 已經開始監聽mysql 的表了。查詢哪些表 就在jdbc.sql 寫sql語句就行了。

現在 手動的 hotel， hotel_account 分別增加一條數據：

 INSERT INTO hotel(hotel_name, photo_url) VALUES("馬二帥酒店","images/madashuai.img");

找到 hotel 的id 

INSERT INTO hotel_account(hotel_id, finance_person) VALUES(15627, "馬二帥");

大約30秒的時候查看es

 

很好馬二帥酒店已經增加進來了。

現在修改下 hotel_account 

UPDATE hotel_account SET finance_person = "馬二帥修改為馬小帥" where id = 1601;

 

修改也能監聽到哦。

OK到此為止，使用logstash-input-jdbc插件增量監聽es就介紹完