Logstash

Logstash由三個元件構造成，分別是input、filter以及output。我們可以吧Logstash三個元件的工作流理解為：input收集資料，filter處理資料，output輸出資料。

不同資料來源：

1、檔案型別：檔案資料來源，我們可以使用input元件的file外掛來獲取資料。file{}外掛有很多的屬性引數。具體內容在下面的程式碼中展示：

input{
    file{
        #path屬性接受的引數是一個數組，其含義是標明需要讀取的檔案位置
        path => [‘pathA’，‘pathB’]
        
 
#表示多就去path路徑下檢視是夠有新的檔案產生。預設是15秒檢查一次。
        discover_interval => 15
        #排除那些檔案，也就是不去讀取那些檔案
        exclude => [‘fileName1’,‘fileNmae2’]
        #被監聽的檔案多久沒更新後斷開連線不在監聽，預設是一個小時。
        close_older => 3600
        #在每次檢查檔案列 表的時候， 如果一個檔案的最後 修改時間 超過這個值， 就忽略這個檔案。 預設一天。
        ignore_older => 86400
        #
 
logstash 每隔多 久檢查一次被監聽檔案狀態（ 是否有更新） ， 預設是 1 秒。
        stat_interval => 1
        #sincedb記錄資料上一次的讀取位置的一個index
        sincedb_path => ’$HOME/. sincedb‘
        #logstash 從什麼 位置開始讀取檔案資料， 預設是結束位置 也可以設定為：beginning 從頭開始
        start_position => ‘beginning’
        #注意：這裡需要提醒大家的是，如果你需要每次都從同開始讀取檔案的話，關設定start_position => beginning是沒有用的，你可以選擇sincedb_path 定義為 /dev/null
 

    }           

}

2、資料庫型別：資料庫型別的資料來源，使用input元件的JDBC外掛jdbc{}。同樣，jdbc{}具有很多屬性，我們在下面的程式碼中作出說明；

input{
    jdbc{
    #jdbc sql server 驅動,各個資料庫都有對應的驅動，需自己下載
    jdbc_driver_library => "/etc/logstash/driver.d/sqljdbc_2.0/enu/sqljdbc4.jar"
    #jdbc class 不同資料庫有不同的 class 配置
    jdbc_driver_class => "com.microsoft.sqlserver.jdbc.SQLServerDriver"
    #配置資料庫連線 ip 和埠，以及資料庫   
    jdbc_connection_string => "jdbc:sqlserver://200.200.0.18:1433;databaseName=test_db"
    #配置資料庫使用者名稱
    jdbc_user =>   
    #配置資料庫密碼
    jdbc_password =>
    #上面這些都不重要，要是這些都看不懂的話，你的老闆估計要考慮換人了。重要的是接下來的內容。
    # 定時器 多久執行一次SQL，預設是一分鐘
    # schedule => 分 時 天 月 年  
    # schedule =>  22     表示每天22點執行一次
    schedule => "  *"
    #是否清除 last_run_metadata_path 的記錄,如果為真那麼每次都相當於從頭開始查詢所有的資料庫記錄
    clean_run => false
    #是否需要記錄某個column 的值,如果 record_last_run 為真,可以自定義我們需要表的欄位名稱，
    #此時該引數就要為 true. 否則預設 track 的是 timestamp 的值.
    use_column_value => true
    #如果 use_column_value 為真,需配置此引數. 這個引數就是資料庫給出的一個欄位名稱。當然該欄位必須是遞增的，可以是 資料庫的資料時間這類的
    tracking_column => create_time
    #是否記錄上次執行結果, 如果為真,將會把上次執行到的 tracking_column 欄位的值記錄下來,儲存到 last_run_metadata_path 指定的檔案中
    record_last_run => true
    #們只需要在 SQL 語句中 WHERE MY_ID > :last_sql_value 即可. 其中 :sql_last_value 取得就是該檔案中的值
    last_run_metadata_path => "/etc/logstash/run_metadata.d/my_info"
    #是否將欄位名稱轉小寫。
    #這裡有個小的提示，如果你這前就處理過一次資料，並且在Kibana中有對應的搜尋需求的話，還是改為true，
    #因為預設是true，並且Kibana是大小寫區分的。準確的說應該是ES大小寫區分
    lowercase_column_names => false
    #你的SQL的位置，當然，你的SQL也可以直接寫在這裡。
    #statement => SELECT * FROM tabeName t WHERE  t.creat_time > :sql_last_value
statement_filepath => "/etc/logstash/statement_file.d/my_info.sql" #資料型別，標明你屬於那一方勢力。單了ES哪裡好給你安排不同的山頭。 type => "my_info" } #注意：外載的SQL檔案就是一個文字檔案就可以了，還有需要注意的是，一個jdbc{}外掛就只能處理一個SQL語句， #如果你有多個SQL需要處理的話，只能在重新建立一個jdbc{}外掛。 }

3.埠

input {
  beats {
    #接受資料埠
    port => 5044
    #資料型別
    type => "logs"
  }
  #這個外掛需要和filebeat進行配很這裡不做多講，到時候結合起來一起介紹。
}