通过logstash将sqlserver的数据同步到ELK

2020-07-27

通过logstash将sqlserver的数据同步到ELK

前提：搭建并启动ELK（可以不包含filebeat）
搭建过程参考：https://www.jianshu.com/p/9e0aefa0fbaa

下载安装JDBC

1.下载jdbc驱动

版本：7.4
下载地址：(网上找不的可以留言跟我拿)
解压路径：D:\Linux\data\ELK-Windows\elk\logstash-7.8.0\Microsoft JDBC Driver 7.4 for SQL Server
然后把解压文件夹下的mssql-jdbc-7.4.1.jre8.jar复制到D:\Linux\data\ELK-Windows\elk\logstash-7.8.0\bin\jdbcconfig（jdbcconfig目录是新建的）

aa.png

（未执行启动命令前文件夹中只包含这两个东东）

2.新增配置文件jdbc.conf

在D:\Linux\data\ELK-Windows\elk\logstash-7.8.0\bin\jdbcconfig目录下创建配置文件jdbc.conf，内容如下：
注意：这个配置文件不能有中文注释，容易报错

input {
    jdbc {
        jdbc_driver_library => "D:\Linux\data\ELK-Windows\elk\logstash-7.8.0\bin\jdbcconfig\mssql-jdbc-7.4.1.jre8.jar"        #数据库驱动路径
        jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"     #驱动名称
                jdbc_connection_string => "jdbc:sqlserver://172.10.0.108:1433;DatabaseName=elkdb;"       #数据库的连接字符串，数据库服务器的ip和端口，数据库名
                jdbc_user => "sa"      #连接数据库的用户
                jdbc_password => "123456"     #连接数据库的密码
                schedule => "* * * * *"     #重复执行导入任务的时间间隔
                jdbc_default_timezone => "Asia/Shanghai"      #默认时区设置
        statement => "SELECT * FROM elktable where elkid > :sql_last_value"     #导入的表（查询SQL，可以过滤数据）
    }
}

output {
    elasticsearch {
        hosts => ["172.10.0.108:9200"]      #要把数据同步到的es的主机ip
        manage_template => false
        index => "elkdb"    #索引名称（在kibana创建索引时以数据库名称为所以查询其下的表数据）
    document_type => "elktable"       #类型名称（相当于数据库表名）
        document_id => "%{elkid}"      #相当于主键
    }
    stdout { codec => rubydebug }      #输出时的字符编码
}

3.安装启动数据库

详细安装过程自行百度
配置用户：sa 密码：123456 开端口1433
在SQL Server 2016配置管理器中启用SQL Server网络配置下MSSQLSERVER的协议中的Named Pipes，然后重启SQL Server（MSSQLSERVER）服务。（重启过程：Windows搜索服务》找到SQL Server（MSSQLSERVER）右键重启）

bb.png

（测试一下能否通过客户端使用账户密码连接数据库）
用MicrosoftSQL Server Management Studio工具连接本地数据库，执行以下操作
右击数据库---新建数据库---填写相关信息

cc.png

新建查询（注意操作的数据库是elkdb）

USE elkdb

SET ANSI_NULLS ON

SET QUOTED_IDENTIFIER ON

CREATE TABLE [dbo].[elktable](
    [elkid] [NVARCHAR](50) NOT NULL,
    [elkname] [NVARCHAR](50) NULL,
    [elkage] [INT] NULL,
    [elksex] [BIT] NULL,
    [elkbirth] [DATETIME] NULL
) ON [PRIMARY]

#给表elktable添加约束（约束名：DF_elktable_elkid），默认elkid是随机的，NEWID()随机函数
ALTER TABLE [dbo].[elktable] ADD CONSTRAINT [DF_elktable_elkid] DEFAULT (NEWID()) FOR [elkid]

ALTER TABLE [dbo].[elktable] ADD CONSTRAINT [DF_elktable_elkbirth] DEFAULT (GETDATE()) FOR [elkbirth]

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

4.执行导入

powershell进入目录logstash的bin目录D:\Linux\data\ELK-Windows\elk\logstash-7.8.0\bin

.\logstash.bat -f .\jdbcconfig\jdbc.conf --path.data=./jdbcconfig

如果输出是这个样子的，那基本上成了

dd.png

5.kibana查数据

主菜单（左上角三条横线）---Dev Tools

ee.png

控制台输入 GET elkdb/_search 并发送请求，如果右边有输出就说明sqlserver---->elk的数据是通的

ff.png

在主菜单----Discover----打开----管理搜索----Kibana（索引模式）----创建索引模式----elkdb*，下一步----时间筛选时段（@timestamp）----主菜单----选择elkdb的索引----有数据亦说明成功

gg.png