在今天的这篇文章中，我们来讲述一下如何把一个 MySQL 的数据库导入到 Elasticsearch 中并对数据进行分析。

安装

在今天的这个练习中，我们需要使用到 MySQL 数据。在做这个练习之前，我假设我们已经安装好自己的 Elasticsearch 及 Kibana。如果没有安装，请参照我如下的文章进行安装：

• 安装 Elasticsearch。请参阅文章 “如何在 Linux，MacOS 及 Windows 上进行安装 Elasticsearch”。把 Elasticsearch 运行起来
• 请参阅文章 “如何在 Linux 及 MacOS 上安装 Elastic 栈中的 Kibana”。把 Kibana 运行起来。

MySQL安装

根据不同的操作系统我们分别对 MySQL 进行安装。我们可以访问网页来对 MySQL 进行安装。等我们安装完我们的 MySQL 后，在我们的 terminal 中，打入如下的命令来检查 MySQL 的版本：

$ /usr/local/mysql/bin/mysql -V
/usr/local/mysql/bin/mysql  Ver 8.0.17 for macos10.14 on x86_64 (MySQL Community Server - GPL)

我们可以使用如下的命令来登录 MySQL：

mysql -u root -p

Logstash 安装

请参阅网址 “如何安装 Elastic 栈中的 Logstash” 来安装我们的 Logstash。在上一步中，我们已经拿到了我们的 mysql 的版本信息。我们需要下载相应的 JDBC connector。针对我的情况，我在地址 MySQL :: Download Connector/J 下载最新的 Connector。下载完这个 Connector后，把这个 connector 存入到 Logstash 安装目录下的如下子目录中。

$ pwd
/Users/liuxg/elastic/installations/logstash-7.3.0
(base) localhost:logstash-7.3.0 liuxg$ ls logstash-core/lib/jars/mysql-connector-java-8.0.17.jar 

logstash-core/lib/jars/mysql-connector-java-8.0.17.jar

这样我们的安装就完成了。

准备练习数据

为了方便，我们可以采用把一个 CSV 文件导入到 MySQL 中的办法来形成一个 MySQL 的数据库。我们可以在如下的地址进行下载：

https://github.com/liu-xiao-guo/sample_csv

在上面的 sample_csv 中，我们会发现一个叫做 SalesJan2009.csv 的文件。我们通过一些 MySQL 的前端工具把这个导入到我们的 MySQL 数据库中。

这样我们的 MySQL 的数据库 data 里含有一个叫做 SalesJan2009 的数据就建立好了。

Logstash 配置

我们可以对 Logstash 做如下的配置 sales.conf:

input {
	jdbc {
       jdbc_connection_string => "jdbc:mysql://localhost:3306/data"
       jdbc_user => "root"
       jdbc_password => "YourMyQLPassword"
       jdbc_validate_connection => true
       jdbc_driver_library => ""
       jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
       parameters => { "Product_id" => "Product1" }
       statement => "SELECT * FROM SalesJan2009 WHERE Product = :Product_id"
    }    
}

filter {
	mutate {
	 	rename => {
        	"longitude" => "[location][lon]"
        	"latitude" => "[location][lat]"
    	}
    }
}

output {
	stdout {
	}

   	elasticsearch {
     	index => "sales"
     	hosts => "localhost:9200"
     	document_type => "_doc"
	} 
}

在这里，我们必须替换 jdbc_user 和 jdbc_password 为自己的 MySQL 账号的用户名及密码。特别值得指出的是 jdbc_driver_library 按 elastic 的文档是可以放入 JDBC 驱动的路径及驱动名称。实践证明如果这个驱动不在 JAVA 的 classpath 里，也是不能被正确地加载。正因为这样的原因，我在上一步里把驱动 mysql-connector-java-8.0.17.jar 放入到 Logstash 的 jar 目录里，所以这里我们就直接填入空字符串。

运行 Logstash 加载数据

接下来我们运行 Logstash 来加载我们的 MySQL 里的数据到 Elasticsearch 中：

sudo ./bin/logstash --debug -f  ~/data/sales.conf 

在这里我们把 sales.conf 置于用户 home 目录下的 data 子目录中。

我们可以在 Kibana 中查看到最新的导入到 Elasticsearch 中的数据：

这里显示在 sales 索引中有 847 个文档。一旦数据进入到我们的 Elastic，我们可以对数据进行分析：

参考：

【1】Jdbc input plugin | Logstash Reference [8.5] | Elastic