使用Logstash工具匯入sqlserver資料到elasticSearch及elk分散式紀錄檔中心

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://sqlserverIP:1433;databaseName=Reptile.NewsLetter"
    jdbc_user => "sa"
    jdbc_password => "密碼"
    #分頁且最大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"
       }
}

<?xml version="1.0" encoding="utf-8" ?>
<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      autoReload="true"
      internalLogLevel="Warn"
      internalLogFile="internal-nlog.txt">

  <extensions>
    <add  assembly="NLog.Web.AspNetCore"/>
  </extensions >
  <variable name="logDirectory" value="${basedir}\logs\"/>
  <!--define various log targets-->
  <targets>
    <!--write logs to file-->
    <!--address 填寫Logstash資料的接收通道-->
    <target xsi:type="Network"
            name="elastic"
            keepConnection="false"
            address ="tcp://my.es.com:50000"
            layout="${longdate}|${logger}|${uppercase:${level}}|${message} ${exception}" />
    />
    <target xsi:type="Null" name="blackhole" />
  </targets>
  <rules>
    <!--All logs, including from Microsoft-->
    <!--<logger name="*" minlevel="Trace" writeTo="allfile" />-->
    <!--Skip Microsoft logs and so log only own logs-->
    <logger name="Microsoft.*" minlevel="Trace" writeTo="blackhole" final="true" />
    <logger name="*" minlevel="Trace" writeTo="elastic" />
  </rules>
</nlog>