利用logstash的logstash-input-jdbc插件实现mysql增量导入ES的介绍

矫情吗；* 2022-06-14 03:59 113阅读 0赞

参考链接：

1， [源码地址，官方介绍][Link 1]

2， [logstash-input-jdbc使用建议][logstash-input-jdbc]

3, [官网介绍例子，使用 logstash-input-jdbc 到 elasticsearch][logstash-input-jdbc _ elasticsearch]

一、安装 elastisearch, logstash

参考我的上一篇链接 ：[centos下 ELK部署文档][centos_ ELK]

二、安装logstash-input-jdbc插件

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

使用ruby语言开发。

下载插件过程中最大的坑是下载插件相关的依赖的时候下不动，因为国内网络的原因，访问不到亚马逊的服务器。

解决办法，改成国内的ruby仓库镜像。此镜像托管于淘宝的阿里云服务器上 ：

[淘宝镜像官网地址][Link 2]

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-input-jdbc 1]

选择对应的版本。我的logstash版本是1.4.0，对应的插件版本是1.0.0

关于插件版本的选择 参考这里：[这是ruby Gemfile所有插件的官方地址，参考logstash-core ，如果你的logstash版本跟他对应上，那么这个插件的版本就是你要的版本][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插件的问题][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][][数据库][MySQL]，我有一张hotel 表， hotel\_account表（此表里有hotel\_id）, 里面无数据。

2，已经启动 elasticsearch . 地址是 [http://192.168.0.45][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官方参考文档][logstash-input-jdbc 2]

重点字段说明：

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

如果一切顺利 应该如图：

![193802_xSum_871381.png][]

现在 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

![193819_gk5d_871381.png][]

很好马二帅酒店已经增加进来了。

现在修改下 hotel\_account

UPDATE hotel_account SET finance_person = "马二帅修改为马小帅" where id = 1601;

![Center][]

修改也能监听到哦。

OK到此为止，使用logstash-input-jdbc插件增量监听es就介绍完咯

[Link 1]: https://github.com/logstash-plugins/logstash-input-jdbc
[logstash-input-jdbc]: https://github.com/elastic/logstash/issues/3429
[logstash-input-jdbc _ elasticsearch]: https://www.elastic.co/blog/logstash-jdbc-input-plugin
[centos_ ELK]: https://tower.im/projects/2eec08a5c51c4161be0ba3bec5ef7444/docs/16dedcf069a8461ca9350a47225dda40/
[Link 2]: https://ruby.taobao.org/
[logstash-input-jdbc 1]: https://github.com/logstash-plugins/logstash-input-jdbc/releases
[ruby Gemfile_logstash-core _logstash]: https://rubygems.org/gems/logstash-input-jdbc
[jdbc]: https://github.com/logstash-plugins/logstash-input-jdbc/issues/38
[MySQL]: http://lib.csdn.net/base/mysql
[http_192.168.0.45]: http://192.168.0.45/
[Java]: http://lib.csdn.net/base/java
[logstash-input-jdbc 2]: https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html
[193802_xSum_871381.png]: /images/20220614/cadbe116cc7a4798adf21113e41fa067.png
[193819_gk5d_871381.png]: /images/20220614/4cebf72923684aa68529603102ab5c2a.png
[Center]: /images/20220614/ec408c61f0324e16b9babc9e54dc15c2.png