Windows Logstash同步 Sqlserver 到Elasticsearch

下載地址:https://artifacts.elastic.co/downloads/logstash/logstash-7.13.2-windows-x86_64.zip
JDBC  https://docs.microsoft.com/en-us/sql/connect/jdbc/download-microsoft-jdbc-driver-for-sql-server?view=sql-server-ver15

input {
stdin {
    }
  jdbc {
     #數據庫驅動所在位置，可以是絕對路徑或者相對路徑
    jdbc_driver_library => "D:\software\Elasticsearch\logstash-7.13.2\bin\jdbcconfig\mssql-jdbc-9.2.1.jre8.jar"
    #驅動類名
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
     #數據庫連接
    jdbc_connection_string => "jdbc:sqlserver://服務器名稱;DatabaseName=CstCRMTest;"
    #用戶
    jdbc_user => "sa"
    #密碼
    jdbc_password => "const-123456"
    #設置定時任務間隔  含義：分、時、天、月、年，全部為*默認含義為每分鐘跑一次任務
    schedule => "* * * * *"
    #sql語句
    statement => "SELECT  [Id]
      ,[QuoteCode]
      ,[QuoteName]
      ,[CustomerId]
      ,[QuotePerson]
      ,[UserId]
      ,[QuotePhone]
      ,[PayType]
      ,[QuoteVailDate]
      ,[LeadTime]
      ,[QuoteDate]
      ,[CompanyPhone]
      ,[CompanyAddress]
      ,[CompanyUrl]
      ,[Remark]
      ,[CreatedTime]
      ,CONVERT (VARCHAR (30),UpdatedTime,25) AS updatedTime
      ,[CreatedUser]
      ,[UpdatedUser]
    FROM [CstCRMTEST].[dbo].[T_Quote]  where  updatedTime>:sql_last_value"
    #sql可執行文件
    #statement_filepath => "路徑"
    #是否開啟記錄上次追蹤的結果，也就是上次更新的時間，這個會記錄到 last_run_metadata_path 的文件
    use_column_value => true
    # 對應字段的類型
    tracking_column_type => "timestamp"
    #如果 use_column_value 為true， 配置本參數，追蹤的 column 名，可以是自增id或者時間
    tracking_column => "updatedTime"
    #是否記錄上次執行結果, 如果record_last_run為真,將會把上次執行到的 tracking_column 字段的值記錄下來,保存到 last_run_metadata_path 指定的文件中
    record_last_run => true
    # 記錄上一次追蹤的結果值，保存文件到對應路徑中
     last_run_metadata_path => "D:\software\Elasticsearch\logstash-7.13.2\bin\jdbcconfig\updatedTime.txt"
      # 索引類型
      #type => "_doc"
      # 數據庫字段名稱大寫轉小寫
      lowercase_column_names => false
     #是否清除 last_run_metadata_path 的記錄,如果為真那么每次都相當于從頭開始查詢所有的數據庫記錄
     #clean_run : 
  }
}
 
 
output {
    elasticsearch {
        # ES的IP地址及端口
        hosts => ["localhost:9200"]
        # 索引名稱 可自定義(只可以小寫)
        index => "quote"
        document_type => "out"
        # 需要關聯的數據庫中有有一個id字段，對應類型中的id
        document_id => "%{Id}"
    
    }
 stdout {
       # codec => json_lines
#設置輸出的格式
   codec => line {
  format => "updatedTime: %{[updatedTime]}"
   }
    }

}