用Logstash中的Jdbcinputplugin将mysql数据导入ES

1. 在CentOS上安装logstash5.5：

参考这篇文章。 只是这文章比较久了，现在有些变化，我还是把过程写下来：

1，下载logstash5.5.1

curl -O https://artifacts.elastic.co/downloads/logstash/logstash-5.5.1.tar.gz

2, 解压tar zxvf logstash-5.5.1.tar.gz

3, 进入logstash-5.5.1目录， 根据网上的帖子，还需要单独安装logstash-input-jdbc, 但是现在都已经集成在5.5.1的安装包i面了

1. 参考官方的plugin说明文档， 在命令行中运行：

[root@VM_27_195_centos logstash-5.5.1]# bin/logstash-plugin list

该命令列出所有已安装的插件

bin/logstash-plugin list --verbose

这个命令列出所有已安装插件的名称和版本，截屏部分如下，你会发现logstash-input-jdbc(4.2.1)已经安装了：

2. 安装ElasticSearch

安装文档
根据安装好后，运行./bin/elasticsearch, 报内存不够，因为我的虚机只有1G的内存，运行’top’命令发现只有150M的可用内存，怎么办呢？修改/usr/local/es/elasticsearch-5.4.3/config目录下的jvm.options文件:

vim jvm.options

修改成128m:

################################################################
## IMPORTANT: JVM heap size
################################################################
##
## You should always set the min and max JVM heap
## size to the same value. For example, to set
## the heap to 4 GB, set:
##
## -Xms4g
## -Xmx4g
##
## See https://www.elastic.co/guide/en/elasticsearch/reference/current/heap-size.html
## for more information
##
################################################################

# Xms represents the initial size of total heap space
# Xmx represents the maximum size of total heap space




-Xms128m
-Xmx128m

ES不能用root用户启动，建立一个新的用户es, 切换到es:

su es

之后在ES的安装目录里执行：

./bin/elasticsearch

可以在命令行里用curl命令查看ES是否已经运行（默认端口9200），命令如下：

curl -XGET 'localhost:9200/?pretty'

系统会显示：

3. 安装Mysql, 情形和这个帖子描述的一样，最后还是装了又卸掉。

安装原生tar.gz包的官方指南
太麻烦，于是安装了MariaDB, 安装教程
配置MariaDB，root的密码是simon, 之后加了一个可以从远程访问的帐号：

MariaDB [(none)]> CREATE USER 'zoujia'@'%' IDENTIFIED BY 'simon';

并授予一切权利：

MariaDB [(none)]>  GRANT ALL PRIVILEGES ON *.* TO 'zoujia'@'%' WITH GRANT OPTION;

这样就能从NaviCat远程连接了。

4. 使用logstash的logstash-input-jdbc插件来将mysql数据库中的数据导入到ES， 根据这个官方说明操作

1. 首先要在logstash目录下新建一个jdbc.conf文件:

input {
    stdin {
    }
    jdbc {
      # mysql jdbc connection string to our backup databse
      jdbc_connection_string => "jdbc:mysql://localhost:3306/test"
      # the user we wish to excute our statement as
      jdbc_user => "simon"
      jdbc_password => "simon"
      # the path to our downloaded jdbc driver
      jdbc_driver_library => "/opt/logstash-5.5.1/mysql-connector-java-5.1.36.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 {
        hosts => ["127.0.0.1:9200"]
        index => "mysql01"
        document_id => "%{id}"    }
    stdout {
        codec => json_lines
    }
}

1. 其中要上传一个’mysql-connector-java-5.1.36.jar’文件到logstash目录下，以对应配置文件中的：

jdbc_driver_library => "/opt/logstash-5.5.1/mysql-connector-java-5.1.36.jar"

1. 在logstash目录下建立一个jdbc.sql文件：

select * from simontable

对应配置文件中：

 statement_filepath => "jdbc.sql"
``` 选项

4. 运行命令：





class="se-preview-section-delimiter">div>

[root@VM_27_195_centos logstash-5.5.1]# bin/logstash -f jdbc.conf

等上大约半分钟（没办法，买的虚机配置很低， 只有1G的内存), 开始工作：

![这里写图片描述](http://img.blog.csdn.net/20170802114003824?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvc2ltb25saW5iaW4=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)

系统会每一秒钟执行一次， 在elasticsearch里面执行：





<div class="se-preview-section-delimiter">div>

[root@VM_27_195_centos ~]# curl http://localhost:9200/mysql01/_search?pretty

会显示已经导入的记录：





<div class="se-preview-section-delimiter">div>

{
“took” : 120,
“timed_out” : false,
“_shards” : {
“total” : 5,
“successful” : 5,
“failed” : 0
},
“hits” : {
“total” : 7,
“max_score” : 1.0,
“hits” : [
{
“_index” : “mysql01”,
“_type” : “jdbc”,
“_id” : “5”,
“_score” : 1.0,
“_source” : {
“@timestamp” : “2017-08-02T03:11:00.220Z”,
“name” : “linzhongyue”,
“@version” : “1”,
“id” : 5,
“type” : “jdbc”,
“age” : 81
}
},
{
“_index” : “mysql01”,
“_type” : “jdbc”,
“_id” : “2”,
“_score” : 1.0,
“_source” : {
“@timestamp” : “2017-08-02T03:11:00.218Z”,
“name” : “wuraorao”,
“@version” : “1”,
“id” : 2,
“type” : “jdbc”,
“age” : 36
}
},
{
“_index” : “mysql01”,
“_type” : “jdbc”,
“_id” : “4”,
“_score” : 1.0,
“_source” : {
“@timestamp” : “2017-08-02T03:11:00.219Z”,
“name” : “zouwangbei”,
“@version” : “1”,
“id” : 4,
“type” : “jdbc”,
“age” : 5
}
},
{
“_index” : “mysql01”,
“_type” : “jdbc”,
“_id” : “6”,
“_score” : 1.0,
“_source” : {
“@timestamp” : “2017-08-02T03:11:00.220Z”,
“name” : “zouyong”,
“@version” : “1”,
“id” : 6,
“type” : “jdbc”,
“age” : 33
}
},
{
“_index” : “mysql01”,
“_type” : “jdbc”,
“_id” : “1”,
“_score” : 1.0,
“_source” : {
“@timestamp” : “2017-08-02T03:11:00.218Z”,
“name” : “simon”,
“@version” : “1”,
“id” : 1,
“type” : “jdbc”,
“age” : 43
}
},
{
“_index” : “mysql01”,
“_type” : “jdbc”,
“_id” : “7”,
“_score” : 1.0,
“_source” : {
“@timestamp” : “2017-08-02T03:11:00.223Z”,
“name” : “hanxiaodong”,
“@version” : “1”,
“id” : 7,
“type” : “jdbc”,
“age” : 44
}
},
{
“_index” : “mysql01”,
“_type” : “jdbc”,
“_id” : “3”,
“_score” : 1.0,
“_source” : {
“@timestamp” : “2017-08-02T03:11:00.219Z”,
“name” : “lingwangnan”,
“@version” : “1”,
“id” : 3,
“type” : “jdbc”,
“age” : 6
}
}
]
}
}

“`

测试结果： 增加和更改MariaDB中的原表，都可以在ES中反馈出来，但如果是删除了一条记录，但是在ES该记录仍然存在，只是他的@timestamp不再更行了，见下图：

第二条记录在MariaDB中已经删除了，你可以看到它的时间比上面的记录早3分钟。

[root@VM_27_195_centos ~]# curl http://localhost:9200/mysql01/_search?pretty

会显示已经导入的记录：

{
  "took" : 120,
  "timed_out" : false,
  "_shards" : {
 "total" : 5,
 "successful" : 5,
 "failed" : 0
 },
  "hits" : {
 "total" : 7,
 "max_score" : 1.0,
 "hits" : [
 {
 "_index" : "mysql01",
 "_type" : "jdbc",
 "_id" : "5",
 "_score" : 1.0,
 "_source" : {
 "@timestamp" : "2017-08-02T03:11:00.220Z",
 "name" : "linzhongyue",
 "@version" : "1",
 "id" : 5,
 "type" : "jdbc",
 "age" : 81
 }
 },
 {
 "_index" : "mysql01",
 "_type" : "jdbc",
 "_id" : "2",
 "_score" : 1.0,
 "_source" : {
 "@timestamp" : "2017-08-02T03:11:00.218Z",
 "name" : "wuraorao",
 "@version" : "1",
 "id" : 2,
 "type" : "jdbc",
 "age" : 36
 }
 },
 {
 "_index" : "mysql01",
 "_type" : "jdbc",
 "_id" : "4",
 "_score" : 1.0,
 "_source" : {
 "@timestamp" : "2017-08-02T03:11:00.219Z",
 "name" : "zouwangbei",
 "@version" : "1",
 "id" : 4,
 "type" : "jdbc",
 "age" : 5
 }
 },
 {
 "_index" : "mysql01",
 "_type" : "jdbc",
 "_id" : "6",
 "_score" : 1.0,
 "_source" : {
 "@timestamp" : "2017-08-02T03:11:00.220Z",
 "name" : "zouyong",
 "@version" : "1",
 "id" : 6,
 "type" : "jdbc",
 "age" : 33
 }
 },
 {
 "_index" : "mysql01",
 "_type" : "jdbc",
 "_id" : "1",
 "_score" : 1.0,
 "_source" : {
 "@timestamp" : "2017-08-02T03:11:00.218Z",
 "name" : "simon",
 "@version" : "1",
 "id" : 1,
 "type" : "jdbc",
 "age" : 43
 }
 },
 {
 "_index" : "mysql01",
 "_type" : "jdbc",
 "_id" : "7",
 "_score" : 1.0,
 "_source" : {
 "@timestamp" : "2017-08-02T03:11:00.223Z",
 "name" : "hanxiaodong",
 "@version" : "1",
 "id" : 7,
 "type" : "jdbc",
 "age" : 44
 }
 },
 {
 "_index" : "mysql01",
 "_type" : "jdbc",
 "_id" : "3",
 "_score" : 1.0,
 "_source" : {
 "@timestamp" : "2017-08-02T03:11:00.219Z",
 "name" : "lingwangnan",
 "@version" : "1",
 "id" : 3,
 "type" : "jdbc",
 "age" : 6
 }
 }
 ]
 }
}

测试结果： 增加和更改MariaDB中的原表，都可以在ES中反馈出来，但如果是删除了一条记录，但是在ES该记录仍然存在，只是他的@timestamp不再更行了，见下图：

第二条记录在MariaDB中已经删除了，你可以看到它的时间比上面的记录早3分钟。