通过Logstash由MySQL和SQLServer向Elasticsearch导入数据

学习整理，仅用参考！

所需软件版本信息

• elasticsearch-6.0.0-beta2.msi
  官方下载链接：https://www.elastic.co/downlo&＃8230;
  请选择对应的操作系统版本
• kibana-6.0.0-beta2-windows-x86_64.zip
  官方下载链接：https://www.elastic.co/downlo&＃8230;
  请选择对应的操作系统版本
• logstash-6.0.0-beta2.zip
  官方下载链接：https://www.elastic.co/downlo&＃8230;
  请选择对应的操作系统版本
• Java-jdk-8u131-windows-x64.exe
  官方下载链接：http://www.oracle.com/technet&＃8230;
  请选择对应的操作系统版本,大版本一致即可
• mysql-connector-java-5.1.44-bin.jar
  官方下载链接：https://dev.mysql.com/downloa&＃8230;
  请选择对应的操作系统版本
• mysql-installer-community-5.7.19.0.msi
  官方下载链接：https://dev.mysql.com/downloa&＃8230;
  请选择对应的操作系统版本
• SQL Server 2016
  官方下载链接：https://www.microsoft.com/en-&＃8230;
• Microsoft JDBC Driver 6.2 for SQL Server
  官方下载链接：https://www.microsoft.com/zh-&＃8230;
• 操作系统：Windows 10 企业版（64位 已激活）

说明：
部分软件官网下载地址的下载速度很慢，如果有类似阿里云服务器的云服务器，
可先在云服务器上进行下载（超快），然后拷贝到本地。
如若网址无法访问，请安装 蓝灯
也可下载我已整理好的，下载链接如下：

• 点击下载ELK6.0三件套（提取密码：zqs1）
• 点击下载ELK5.5.1三件套（提取密码：5o89）
• 点击下载Java-jdk-8u131-windows-x64.exe（提取密码：3bg5）
• 点击下载mysql-installer-community-5.7.19.0.msi（提取密码：05e1）
• 点击下载mysql-connector-java-5.1.44-bin.jar（提取密码：069u）
• 点击下载SQL Server 2016（提取密码：1pf9）
• 点击下载Microsoft JDBC Driver 6.2 for SQL Server（提取密码：0hs1）
• 点击下载配置文件（提取密码：2n4p）

说明：
我已在本地“hosts”（C:WindowsSystem32driversetchosts）文件中添加“127.0.0.1 es”，
以下内容中的“es”请自行替换为“localhost”。

安装ELK

1. 安装 elasticsearch-6.0.0-beta2.msi
   注意安装路径，中间不要有中文和空格，这里我选择安装在C:\Elastic\Elasticsearch；
   其他三个文件（lib、data、log）也放在C:\Elastic\Elasticsearch下（可通过勾选框一键修改）；
   不安装其他插件，如x-pack（安装很慢，还需要配置以及license，后期可以加进去）；
   在浏览器中访问“es:9200”，跳出类似如下内容说明正常（服务已正常启动）：

1. 将kibana-6.0.0-beta2-windows-x86_64.zip和logstash-6.0.0-beta2.zip解压到C:\Elastic下

2. 创建kibana的后台启动文件
   可直接下载上方提供的配置文件，或自行在C:\Elastic\kibana-6.0.0-beta2-windows-x86_64\bin下创建RunKibana.vbs，编辑内容如下：

   Set ws = CreateObject("Wscript.Shell")
ws.run "cmd /c kibana.bat",vbhide


   保存文件后双击“RunKibana.vbs”运行，然后在浏览器中访问“es:5601”。
   正常跳出kibana的界面且没有错误提示说明kibana服务已正常启动（如有问题，请自行谷歌解决问题，然后继续下面的操作）。

安装Java并配置环境变量

JAVA_HOME : C:\Program Files\Java\jdk1.8.0_131（请根据自己的安装路径进行替换）；
CLASSPATH : .;%JAVA_HOME%\lib;%JAVA_HOME%\lib\d
Path : %JAVA_HOME%\bin;%JAVA_HOME%\jre\bin
具体步骤可参照百度经验。

安装MySQL

根据自身需求进行安装，记住用户名和密码。我这里设置为“root”，密码“123qweASD”。
创建数据库“forelk”，并在“forelk”下创建表“elktable”，之后插入数据：

CREATE DATABASE `forelk` /*!40100 DEFAULT CHARACTER SET utf8 COLLATE utf8_bin */;
CREATE TABLE `elktable` (
`elkid` int(11) NOT NULL,
`elkname` varchar(45) COLLATE utf8_bin DEFAULT NULL,
`elkage` int(11) DEFAULT NULL,
`elksex` tinyint(4) DEFAULT NULL,
`elkbirth` date DEFAULT NULL,
PRIMARY KEY (`elkid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin;
INSERT INTO `forelk`.`elktable`
(`elkid`,
`elkname`,
`elkage`,
`elksex`,
`elkbirth`)
VALUES
(111,
aa,
11,
1,
2006);
INSERT INTO `forelk`.`elktable`
(`elkid`,
`elkname`,
`elkage`,
`elksex`,
`elkbirth`)
VALUES
(222,
bb,
22,
0,
1995);

准备MySQL的驱动

在C:\Elastic\logstash-6.0.0-beta2\lib下新建mysqldriver文件夹，并将mysql-connector-java-5.1.44-bin.jar拷贝到mysqldriver文件夹中。

准备Logstash的配置文件

在C:\Elastic\logstash-6.0.0-beta2\config下新建mysql.conf文件，编辑内容如下：

input {
jdbc {
jdbc_driver_library => "C:/Elastic/logstash-6.0.0-beta2/lib/mysqldriver/mysql-connector-java-5.1.44-bin.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://es:3306/forelk?autoRecOnnect=true&useSSL=false"
jdbc_user => "root"
jdbc_password => "123qweASD"
schedule => "* * * * *"
jdbc_default_timezOne=> "Asia/Shanghai"
statement => "SELECT * FROM elktable;"
}
}
output {
elasticsearch {
index => "elkdb"
document_type => "elktable"
document_id => "%{elkid}"
hosts => ["es:9200"]
}
}

参数说明：

• jdbc_driver_library：

数据库驱动路径，这里我填写的是绝对路径，可自行尝试相对路径；

• jdbc_driver_class：

驱动名称；

• jdbc_connection_string：
  数据库的连接字符串；
  forelk为数据库名；
  ?autoRecOnnect=true&useSSL=false自动重连并禁用SSL；
• jdbc_user：

数据库用户名；

• jdbc_password：

数据库密码；

• schedule：

重复执行导入任务的时间间隔；

• jdbc_default_timezone：

默认时区设置；

• statement：

导入的表（查询SQL，可以过滤数据）

• index:

索引名称（类似数据库名称）；

• document_type：

类型名称（类似数据库表名）；

• document_id：

类似主键；

• hosts：

要导入到的Elasticsearch所在的主机；

执行导入

在C:\Elastic\logstash-6.0.0-beta2下Shift+鼠标右键，选择在此处打开命令窗口(W)；
执行bin\logstash -f config\mysql.conf，执行结果如下：

执行查询

在Kibana的Dev Tools下执行GET elkdb/_search命令，
右侧的结果显示类似下图（我的MySQL表中插入了四条数据，所以这里导入的也是四条），
则说明导入成功。

以上为从MySQL导入数据到Elasticsearch的过程

从SQL Server导入数据到Elasticsearch的过程与上方类似。

安装SQL Server 2016

自动谷歌/百度安装，记住用户名和密码。我这里的用户名为sa，密码为123qweASD。
注意防火墙关闭或添加入站规则（开通1433端口）。
在SQL Server 2016配置管理器中启用SQL Server网络配置下MSSQLSERVER的协议中的Named Pipes，然后重启SQL Server（MSSQLSERVER）服务。
确认通过客户端可以通过用户名和密码正常连接数据库。
执行下面的脚本，创建DB并插入数据：

USE [master]
GO
CREATE DATABASE [elkdb]
USE [elkdb]
GO
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE TABLE [dbo].[elktable](
[elkid] [NVARCHAR](50) NOT NULL,
[elkname] [NVARCHAR](50) NULL,
[elkage] [INT] NULL,
[elksex] [BIT] NULL,
[elkbirth] [DATETIME] NULL
) ON [PRIMARY]
GO
ALTER TABLE [dbo].[elktable] ADD CONSTRAINT [DF_elktable_elkid] DEFAULT (NEWID()) FOR [elkid]
GO
ALTER TABLE [dbo].[elktable] ADD CONSTRAINT [DF_elktable_elkbirth] DEFAULT (GETDATE()) FOR [elkbirth]
GO
INSERT [dbo].[elktable] ([elkid], [elkname], [elkage], [elksex], [elkbirth]) VALUES (N'B0048C6E-05AD-4685-861F-E8F4A861D3AB', N'ssa', 123, 0, CAST(N'2017-09-08T09:39:12.567' AS DateTime))
GO
INSERT [dbo].[elktable] ([elkid], [elkname], [elkage], [elksex], [elkbirth]) VALUES (N'EAC8CC17-4E66-4C1C-964B-36D3CD875BDD', N'ssb', 456, 1, CAST(N'2017-09-08T09:39:23.090' AS DateTime))
GO
INSERT [dbo].[elktable] ([elkid], [elkname], [elkage], [elksex], [elkbirth]) VALUES (N'D672806D-6C4F-4CCB-9DCC-434A7ECDA083', N'ssc', NULL, 0, CAST(N'2017-09-08T09:39:33.050' AS DateTime))
GO
INSERT [dbo].[elktable] ([elkid], [elkname], [elkage], [elksex], [elkbirth]) VALUES (N'20DDA41B-A679-4627-B6ED-1D96AF953CF7', N'ssd', 789, NULL, CAST(N'2017-09-08T09:39:41.467' AS DateTime))
GO
INSERT [dbo].[elktable] ([elkid], [elkname], [elkage], [elksex], [elkbirth]) VALUES (N'6F9F8CD8-FCFB-4D88-B191-40DFD360C8C3', NULL, 135, NULL, CAST(N'2017-09-08T09:39:57.353' AS DateTime))
GO

准备SQL Server的驱动

与MySQL类似，在C:/Elastic/logstash-6.0.0-beta2/lib下创建sqlserverdriver文件夹，并将mssql-jdbc-6.2.1.jre8.jar拷贝到该文件夹下。

准备SQL Server的配置文件

在C:/Elastic/logstash-6.0.0-beta2/config下创建sqlserver.conf，编辑内容：

input {
jdbc {
jdbc_driver_library => "C:/Elastic/logstash-6.0.0-beta2/lib/sqlserverdriver/mssql-jdbc-6.2.1.jre8.jar"
jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
jdbc_connection_string => "jdbc:sqlserver://es:1433;databaseName=elkdb;"
jdbc_user => "sa"
jdbc_password => "123qweASD"
schedule => "* * * * *"
jdbc_default_timezOne=> "Asia/Shanghai"
statement => "SELECT * FROM elktable"
}
}
output {
elasticsearch {
index => "sselkdb"
document_type => "sselktable"
document_id => "%{elkid}"
hosts => ["es:9200"]
}
}

参数说明同MySQL配置文件中的说明一致。

执行导入

在C:\Elastic\logstash-6.0.0-beta2下Shift+鼠标右键，选择在此处打开命令窗口(W)；
执行bin\logstash -f config\sqlserver.conf，执行结果类似MySQL。

执行查询

在Kibana的Dev Tools下执行GET sselkdb/_search命令，
查询结果类似MySQL。

参考文档

• Connection URL Sample
• Using the JDBC Driver
• Configuring Logstash
• Logstash Configuration Examples
• 利用Logstash插件进行Elasticsearch与Mysql的数据
• Warning about SSL connection when connecting to MySQL database
• MySQL到Elasticsearch的同步之路

声明

如需转载，请注明原文链接：https://segmentfault.com/a/11&＃8230;！
如有疑问，请发邮件到ably@vip.163.com，Thanks♪(･ω･)ﾉ