Elasticsearch系列(九)----使用Logstash-input-jdbc同步数据库中的数据到ES

我不是女神ヾ 2022-06-09 12:26 281阅读 0赞

**1、数据同步方式**

全量同步与增量同步  
  
全量同步是指全部将数据同步到es，通常是刚建立es，第一次同步时使用。增量同步是指将后续的更新、插入记录同步到es。

**2、常用的一些ES同步方法**

1）、[elasticsearch-jdbc][]: 严格意义上它已经不是第三方插件。已经成为独立的第三方工具。不支持5.5.1。。。  
2）、elasticsearch-river-mysql插件:  [https://github.com/scharron/elasticsearch-river-mysql ][https_github.com_scharron_elasticsearch-river-mysql]  
3）、go-mysql-elasticsearch(国内作者siddontang):  [ https://github.com/siddontang/go-mysql-elasticsearch][https_github.com_siddontang_go-mysql-elasticsearch]  
4）、python-mysql-replication:  github地址 [https://github.com/noplay/python-mysql-replication][https_github.com_noplay_python-mysql-replication]  
5）、MySQL Binlog:  通过 MySQL binlog 将 MySQL 的数据同步给 ES， 只能使用 row 模式的 binlog。  
6）、Logstash-input-jdbc:  github地址 [https://github.com/logstash-plugins/logstash-input-jdbc][https_github.com_logstash-plugins_logstash-input-jdbc]

**3、Logstash-input-jdbc安装**

由于我用的ES版本是5.5.1,elasticsearch-jdbc不支持，只支持2.3.4，这就尴尬了。

![20170906101751478][]

所用这里用Logstash-input-jdbc来同步数据,logstash-input-jdbc插件是logstash 的一个个插件,使用ruby语言开发。所以要先安装ruby,也是为了好使用ruby中的gem安装插件，下载地址:[https://rubyinstaller.org/downloads/][https_rubyinstaller.org_downloads]

![20170906102700568][]

下载下来之后，进行安装

![20170906102919896][]

![20170906102926011][]

安装好之后试下是否安装成功,打开CMD输入:

![20170906103501871][]

OK,然后修改gem的源,使用以下命令查看gem源

gem sources -l

![20170906103645159][]

删除默认的源

gem sources --remove https://rubygems.org/

![20170906103827257][]

添加新的源

gem sources -a http://gems.ruby-china.org/
    gem sources -l

![20170906104110758][]

更改成功,还的修改Gemfile的数据源地址。步骤如下：

gem install bundler
    bundle config mirror.https://rubygems.org https://gems.ruby-china.org

![20170906104326504][]

然后就是安装logstash-input-jdbc,在logstash-5.5.1/bin目录下

![20170906104939997][]

执行安装命令

.\logstash-plugin.bat install logstash-input-jdbc

静等一会儿，成功之后提示如下

![20170906105654306][]

**4、Logstash-input-jdbc使用**

官方文档地址

[https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html][https_www.elastic.co_guide_en_logstash_current_plugins-inputs-jdbc.html]

首先在bin目录下新建一个mysql目录,里面包含jdbc.conf,jdbc.sql文件,加入mysql的驱动

![20170906105941182][]

**A)全量同步**

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
        }
    }

各数据库对应的链接如下:

## MySQL数据库 ##

Driver ="path/to/jdbc-drivers/mysql-connector-java-5.1.35-bin.jar"   //驱动程序  
Class  ="com.mysql.jdbc.Driver";   
URL  ="jdbc:mysql://localhost:3306/db\_name";                           //连接的URL,db\_name为数据库名

## SQL server数据库 ##

Driver ="path/to/jdbc-drivers/sqljdbc4.jar"  
Class  ="com.microsoft.jdbc.sqlserver.SQLServerDriver";  
URL   ="jdbc:microsoft:sqlserver://localhost:1433;DatabaseName=db\_name";     //db\_name为数据库名

## Oracle数据库 ##

Driver ="path/to/jdbc-drivers/ojdbc6-12.1.0.2.jar"  
Class  ="oracle.jdbc.driver.OracleDriver";  
URL   ="jdbc:oracle:thin:@loaclhost:1521:orcl";     //orcl为数据库的SID

## DB2数据库 ##

//连接具有DB2客户端的Provider实例   
Driver ="path/to/jdbc-drivers/jt400.jar"  
Class  ="com.ibm.db2.jdbc.app.DB2.Driver";   
URL   ="jdbc:db2://localhost:5000/db\_name";     //db\_name为数据可名

## PostgreSQL数据库 ##

Driver ="path/to/jdbc-drivers/postgresql-9.4.1201.jdbc4.jar"  
Class  ="org.postgresql.Driver";            //连接数据库的方法   
URL   ="jdbc:postgresql://localhost/db\_name";      //db\_name为数据可名

jdbc.sql配置如下:

select * from person

就一条查询语句对应的表数据如下:

![20170906115708604][]

先启动ES，然后通过sense创建article索引

UT http://localhost:9200/article

![20170906115204474][]

然后通过以下命令启动logstash

.\logstash.bat -f  .\mysql\jdbc.conf

![20170906114818913][]

过一会他就会自动的往ES里添加数据,输出的日志如下:

![20170906115442629][]

执行了SQL查询。查看下article索引会发现多出来了很多文档

![20170906115901595][]

我们在数据库增加一条数据,看他是否自动同步到ES中

![20170906120122281][]

静等一会，发现logstash的日志

![20170906120308256][]

查询了一篇,ES中的数据会多出刚刚插入的那条

![20170906120447202][]

**B)增量同步**

下面使用**增量**来新增数据,需要在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"
    
          #处理中文乱码问题
          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 > :last_sql_value 即可. 其中 :last_sql_value 取得就是该文件中的值(10000).
    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"

这里使用webmagic爬虫来爬取数据，导入到数据库中，先运行爬虫，爬取一些数据

![20170906142656768][]

这里爬取到了277条，然后启动logstash,通过logstash导入到ES中去

![20170906142821660][]

打开mysql目录下的station\_parameter.txt文件

![20170906142904984][]

![20170906142946721][]

这个文件里记录上次执行到的 tracking\_column 字段的值,比如上次数据库有 10000 条记录,查询完后该文件中就会有数字 10000 这样的记录,下次执行 SQL 查询可以从 10001 条处开始,我们只需要在 SQL 语句中 WHERE MY\_ID > :last\_sql\_value 即可. 其中 :last\_sql\_value 取得就是该文件中的值。

然后开启爬虫,爬取数据,往数据库里插，logstash会自动的识别到更新，然后导入到ES中！！

[elasticsearch-jdbc]: https://github.com/jprante/elasticsearch-jdbc
[https_github.com_scharron_elasticsearch-river-mysql]: https://github.com/scharron/elasticsearch-river-mysql
[https_github.com_siddontang_go-mysql-elasticsearch]: https://github.com/siddontang/go-mysql-elasticsearch
[https_github.com_noplay_python-mysql-replication]: https://github.com/noplay/python-mysql-replication
[https_github.com_logstash-plugins_logstash-input-jdbc]: https://github.com/logstash-plugins/logstash-input-jdbc
[20170906101751478]: /images/20220609/51fe907fadca430ebec691bae2e6229c.png
[https_rubyinstaller.org_downloads]: https://rubyinstaller.org/downloads/
[20170906102700568]: /images/20220609/58b2c3f4bb2e41e896d26fd1531529d1.png
[20170906102919896]: /images/20220609/e5757dc5d9f541238c6b3e86d3f6eb1f.png
[20170906102926011]: /images/20220609/38394bd554df4f6096eab49c4717438a.png
[20170906103501871]: /images/20220609/95e4f5d5d835478783a166376b9858fe.png
[20170906103645159]: /images/20220609/d48e6fb797bc405cab006c1ce98cd450.png
[20170906103827257]: /images/20220609/b58ec9cf0d52463b93ee01e2ec4d469a.png
[20170906104110758]: /images/20220609/b92411bcce8b46699f800602ad93e660.png
[20170906104326504]: /images/20220609/d6be7667d9874573821867745fd300a7.png
[20170906104939997]: /images/20220609/4e30136cbf2343608a9273d346e9ecc9.png
[20170906105654306]: /images/20220609/9bebd65469154293b91eda65d9aa4aed.png
[https_www.elastic.co_guide_en_logstash_current_plugins-inputs-jdbc.html]: https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html
[20170906105941182]: /images/20220609/11486f8dd8c34ef484cff38f6a677e43.png
[20170906115708604]: /images/20220609/bd9488e89de240bb9055ebc60813ed7d.png
[20170906115204474]: /images/20220609/b43907aa5b27488688e9bbdcd63ec0ba.png
[20170906114818913]: /images/20220609/45e87ea060754429ac9980f56d37d38f.png
[20170906115442629]: /images/20220609/a4af0082182d41e8bd68c4646fbbbcd4.png
[20170906115901595]: /images/20220609/5fe4b8f4bdb64cbfbfcac7223b780d6a.png
[20170906120122281]: /images/20220609/f3f7a294b31642509659a472b54cc23e.png
[20170906120308256]: /images/20220609/1b60d39bfd0545f1b66fbd111ef8c30a.png
[20170906120447202]: /images/20220609/a5c095ee837d4379ba8321bf8bbfafe5.png
[20170906142656768]: /images/20220609/1961df731d214110a4a8e62d439a47ec.png
[20170906142821660]: /images/20220609/b1216a0b6e094630ace6e9bbbc8d7fbd.png
[20170906142904984]: /images/20220609/94cd3f5e90a04ff69219c6d9852c9363.png
[20170906142946721]: /images/20220609/eefece7c18ac48b295887faa403ab044.png