Logstash在ELK這一整套解決方案中作為資料採集終端，支援對接Kafka、資料庫（MySQL、Oracle）、檔案等等。

而在Logstash內部的資料流轉，主要經過三個環節：input -> filter -> output，顧名思義就是輸入、過濾（處理）以及輸出。接下來通過一個實際的操作案例來感受一下Logstash內部的資料流轉過程。

首先在官網下載Logstash的執行包，https://www.elastic.co/fr/downloads/logstash，由於我本地是Windows環境，因此選擇下載Windows對應的包：

將下載完畢的壓縮包logstash-7.14.1-windows-x86_64.zip進行解壓之後進入bin目錄並建立logstash.conf檔案用於對input、filter以及output進行相應的配置：

接下來在配置檔案中進行響應的配置，直接給出完整的例子並在後文進行詳解：

input {
    file {
        path => "D:/logstash-7.14.1/test-log/test.log"
        start_position => beginning
    }
}

filter {    
    grok {                        
        match => { "message" => "(?<Time>[0-9]{6}\.[0-9]{3})\[(?<LogLevel>\d)\]\[(?<ThreadNo>[0-9]*)\].*(?<Tag>FindResponseByPage)\[fund\_account\=(?<FundAccount>[0-9]*)" }
    }
    
    if 
 
[Tag] != "FindResponseByPage" {
        drop {}
    }
}

output {
    file {
        path => "D:/logstash-7.14.1/test-log/logstash.log"
        codec => line { 
            format => "Time: %{Time}, LogLevel: %{LogLevel}, ThreadNo: %{ThreadNo}, Tag: %{Tag}, FundAccount: %{FundAccount}"
        }
    }
}
 

input 中新增 file 項表示通過檔案進行輸入，path 為檔案的絕對路徑（如果配置非絕對路徑，Logstash會給出報錯提示），start_position => beginning 表示每次從檔案頭開始讀取。

filter中則進行對輸入資料的相關處理進行配置（filter可以不配置，效果是原樣輸出）。

grok是Logstash的核心外掛之一，可以根據配置的表示式進行資料篩選並存入指定的變數名中。grok提供了一系列標準的匹配模板，不過由於grok底層也是基於正則表示式，因此也可以直接輸入正則表示式進行匹配。

這裡搬運一個官網上的例子：

55.3.244.1 GET /index.html 15824 0.043

針對上述格式的資料，可以通過grok提供的標準表示式模板進行匹配：

%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

示例中，IP為預定義的模板，client為準備存入的變數名。很容易看出，grok的標準語法如下:

%{SYNTAX:SEMANTIC}

SYNTAX：匹配值的型別，例如15824可以用NUMBER型別所匹配，55.3.244.1可以使用IP型別匹配。
SEMANTIC：儲存該值的一個變數名，例如GET可能代表的是REST請求中指定的方法，那麼用method來進行儲存。

當然，像上方的配置檔案示例中一樣，通過原生的正則表示式來進行匹配也可以，不使用預定義模板的格式如下：

(?<field_name>the pattern here)

其中field_name表示儲存匹配到內容的變數名，後面部分則是表示式，例如：

093124.597[0][30100]

對應的匹配表示式為：

?<Time>[0-9]{6}\.[0-9]{3})\[(?<LogLevel>\d)\]\[(?<ThreadNo>[0-9]*)\]

根據匹配到的值可以通過drop進行一個篩選，例如只需要LogLevel值為error的資料，可以在filter中新增如下配置：

if [LogLevel] != "error" {
    drop {}
}

output這裡同樣配置了file，也就是將輸入的資料經過處理後輸出到另一個指定檔案中，path為輸出檔案的絕對路徑，format則指定了輸出的格式。

接下來啟動Logstash來觀察一下效果，在bin目錄開啟命令列並輸入啟動命令：

再看一下輸入檔案和輸出檔案中的效果，首先是輸入檔案test.log：

接下來是輸出檔案logstash.log，成功達到了了欄位匹配以及篩選的效果：

最後補充一點，Logstash的外掛非常多，配置項也非常多。建議直接在官網進行學習，再介紹下如何閱讀官網的配置詳解：

導航欄右側有input、filter、output等外掛的欄目，通過點選導航欄或者左側正文部分的具體配置項，可以點選進一步檢視配置詳情：