使用Logstash工具导入sqlserver数据到elasticSearch及elk分布式日志中心

首先记下这个笔记，Logstash工具导入sqlserver数据到elasticSearch。

因为logstash使用java写的，我本地开发是win11，所以javade jdk必须要安装。具体安装不介绍了，就是网上下个java8，不要去官网要账号什么的,不是java开发不太折腾，目前只用java8，记得JAVA_HOME配置一下环境变量。

java version "1.8.0_66"

Java(TM) SE Runtime Environment (build 1.8.0_66-b18)
Java HotSpot(TM) 64-Bit Server VM (build 25.66-b18, mixed mode)

从sqlserver拉取数据需要用到sqlserver 的jdbc驱动，我这里是官网下载的sqljdbc_6.2.2.1_enu,没用最新版本，个人认为java8不太合适什么都最新的。

logstash使用的是7.0.0的版本，这里遇到几个问题。

换了几个版本，7.0一下的起不来，8，0以上的也是起不来。

1.存放的目录不能有空格，2.（我的坑）不该去github上下载版本，发现跑不起来。3.版本8.0以上的需要java11支持。

有了这两个包(都是java的)后面就是配置了。首先我在ogstash-7.0.0\lib下面新建文件夹sqlserverdriver,去sqljdbc_6.2.2.1_enu\sqljdbc_6.2\enu文件夹下面拷贝了mssql-jdbc-6.2.2.jre8.jar文件过来

下面就是操作logstash文件夹下面config下面文件了。首先在jvm.options最下面加上权限。

-Djava.library.path=E:\sqljdbc_6.2.2.1_enu\sqljdbc_6.2\enu\auth\x64

在config文件夹下面新建文件sqlserver_update.conf文件,拷贝下面的内容，每行基本有注释就详不说。

input {
  jdbc {
    jdbc_driver_library=>"E:\sqljdbc_6.2.2.1_enu\sqljdbc_6.2\enu/mssql-jdbc-6.2.2.jre8.jar"
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    jdbc_connection_string => "jdbc:sqlserver://119.8.41.84:1433;databaseName=Reptile.NewsLetter"
    jdbc_user => "sa"
    jdbc_password => "IDCmC@0755@971"
    #分页且最大5万次
    jdbc_paging_enabled => "true"
    jdbc_page_size => "50000"
    #时区按照东八
     jdbc_default_timezone =>"Asia/Shanghai"
     last_run_metadata_path => "E:\logstash-7.0.0\config\last_value_meta.txt"
     #启用追踪，则需要指定tracking_column，默认是timestamp()
     use_column_value => true
     # 如果 use_column_value 为真,需配置此参数. track 的数据库 column 名,该 column 必须是递增的. 一般是主键
     tracking_column => id
     #追踪字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型()
     tracking_column_type => numeric    
     #是否记录上次执行结果, 如果为真,将会把上次执行到的 tracking_column 字段的值记录下来,保存到 last_run_metadata_path 指定的文件中
     record_last_run => true
     #statement_filepath => "E:\logstash-7.0.0\config\update.sql" sql可放到独立文件里面去
     #表里有时间也有时间戳 都可用
     statement => "SELECT * FROM [Reptile.NewsLetter].[dbo].[LivesItems] where id > :sql_last_value "
     schedule => "* * * * *"
     #是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
     clean_run => false 
     #是否将 column 名称转小写
     lowercase_column_names => false
  }
}
output {
   elasticsearch {
        hosts => ["http://my.es.com:9200"]   
        index => "nl_livesitem"
         user => "elastic"
         password => "changeme"
       }
}