當前位置：首頁 > 编程资源 > 综合教程 >内容正文

综合教程

Logstash的简单介绍

發布時間：2024/6/21 综合教程 23 生活家

生活随笔收集整理的這篇文章主要介紹了 Logstash的简单介绍小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

logstash的簡介

logstash的概念：是一款開源的數據收集引擎，具有實時管道處理能力。
logstash具有200多個插件，可以接受各種各樣的數據（如日志、網絡請求、關系型數據庫、傳感器或物聯網等等）
Logstash工作過程：
Logstash 就像管道符一樣，讀取輸入數據，然后處理過濾數據，最后輸出到目的地（數據在線程之間以事件的形式流傳）
logstash數據處理過程包括三個部分：input、filter、output
input和output部分可以使用codecs對數據格式進行處理
Input,Filter,Output和codec：都是以插件形式存在
用戶可以通過設置pipeline配置文件，設置符合需求的input、filter、outer、codecs插件實現對指定數據的采集、處理和輸出功能

logstash的三大部分的介紹：
input：從數據源獲取數據。不同的數據源使用不同的插件。常用的插件有：file、jdbc、Redis、syslog、beats 、http等
filter：用于處理數據。對從數據源獲取到的數據按照需求進行處理（如：解析數據、刪除字段、類型轉換等）。常用的組件有：date、grok、dessect、mutate、json、geoip、ruby等
output：用于將數據輸出到目的地。不同的目的地使用不同的插件。常用的插件有：elaticsearch、file、graphite、statsd等
codec：用于對數據進行編碼解碼。不是一個單獨的流程，是用于input和output部分對數據進行編解碼作用。常見的組件有：json、multiline等
logstash不是一個input-filter-output的數據流，而是一個 input | decode | filter | encode | output 的數據流。

logstash將數據轉換為事件時候，會給事件添加一些額外的信息。下面介紹幾個常見的額為信息：
@timestamp：用來標記事件的發生時間
host：標記事件發生地址
type：標記事件的唯一類型（input和output部分都可以配置多個不同的插件，每個插件可以用type來唯一標記這個插件，可以實現對不同的插件進行不同的處理）
tags：標記事件的某方面屬性。這是一個數組，一個事件可以有多個標簽

Logstash中的數據類型：
bool：use_column_value => true
string：jdbc_driver_class => "com.mysql.jdbc.Driver"
number：jdbc_page_size => 50000
array：hosts =>["192.168.57.100:9200","192.168.57.101:9200","192.168.57.102:9200"]
hash：options =>{key1 =>value1,key2 =>value2}

logastah中的邏輯運算符：
相等關系：==、!=、<、>、<=、>=
正則關系：=~（匹配正則）、!~(不匹配正則）
包含關系：in、not in
布爾操作：and(與）、or（或）、nand（非與）、xor（非或）
一元運算符：!（取反）、()（復合表達式）、!() （對復合表達式結果取反）

三大模塊插件的簡單介紹

input

標準輸入插件：表示從控制臺輸入

input {stdin { }}

file插件：表示從文件讀取數據(如日志文件)

input {
    file {
        # 要導入的文件的位置，
        path => "/var/lib/mysql/slow.log"            
        # 從文件開始的位置開始讀,end表示從結尾開始讀
        start_position => "beginning"
        # 多久之內沒修改過的文件不讀取，0為無限制，單位為秒
        ignore_older => 0  
        # 記錄文件上次讀取位置，輸出到null表示每次都從文件首行開始解析
        sincedb_path => "/dev/null"
        # type字段，可表明導入的日志類型
        type => "mysql-slow"
    }
}

filter

Filter模塊是logstash功能強大的主要原因，可以對Logstash Event進行豐富的處理（比如解析數據、刪除字段、類型轉換等等）
常見的Filter插件如下：

插件名稱	說明
date	日期解析
grok	正則匹配解析
dessect	分隔符解析
mutate	對字段做處理，比如刪除、重命名、替換
json	按照Json格式解析字段內容到指定字段中
geoip	將ip地址進行解析獲取地域信息(包括國別，省市，經緯度等)
ruby	利用ruby代碼來動態修改logstash event

Date插件：可以將日期字符串解析為日期類型，然后替換@timestamp字段或者指定成其他字段(如果不指定為@timestamp字段,會使用系統的時間作為@timestamp的值)
將input中的event_date字段的字符串解析日期格式，將解析后的的數據作為@timestamp的值；event_date的日期字符串格式需要和后面的yyyy-MM-dd HH:mm:ss對應
如：2020/12/31 12:23:32 對應的格式為match =>["event_date","yyyy/MM/dd HH:mm:ss"]

#將event_date字符串解析為日期類型，并將解析后的值作為@timestamp的值
filter {
    date{
            match =>["event_date","yyyy-MM-dd HH:mm:ss"]
            target =>"@timestamp"
    }
}

Grok插件：基于正則解析數據
grok是filter最重要的插件，grok使用正則表達式來生成grok語法，將input的數據解析成需要的數據格式
grok中常用patterns的路徑：logstash_filevendorundlejrubyx.xgemslogstash-patterns-core-x.x.xpatternsgrok-patterns
grok語法：%{SYNTAX:SEMANTIC}
SYNTAX：grok pattern的名稱
SEMANTIC：賦值字段名稱(將匹配到的值賦值給SEMANTIC這個字段)
如：%{NUMBER:field}：可以匹配數值類型，但是grok匹配出的內容都是字符串類型，可以通過在最后指定為int或者float來強轉類型：%{NUMBER:field:int}
實例：控制臺輸入：hello 123456 666666

filter{
    grok{
        match => {
            "message" => "hello %{NUMBER:num_file:int} %{NUMBER:number_file}"
        }
    }
}

自定義正則表達式：
例子：將匹配到的值賦值給num_file這個字段：(?<num_file>[0-9]{6})
控制臺輸入：hello 123456 666666

filter{
    grok{
        match => {
            "message" => "hello (?<num_file>[0-9]{6}) %{NUMBER:number_file}"
        }
    }
}

自定義grok pattern：通過pattern_definitions參數，以鍵值對的方式定義pattern名稱和內容（也可以通過pattern_dir參數，以文件的形式讀取pattern）
例子：定義模板的名稱為pattern_test，模板的內容是正則匹配8位的數字或者字母；并且使用了自定義的pattern
控制臺輸入：hello 1234abcd

filter{
    grok{
        match => {
            "message" => "%{pattern_test:number_file}"
        }
        pattern_definitions => {
            "pattern_test" => "hello [a-z0-9]{8}"
        }
    }
}

Dissect插件：基于分隔符原理解析數據
解決grok解析時消耗過多cpu資源的問題
語法簡單，能處理的場景比較有限
它只能處理格式相似，且有分隔符的字符串
語法如下：
%{}里面是字段；字段名稱可以自己定義
兩個%{}之間是分隔符

例子：把日期和時間解析到同一個字段datetime_flie
控制臺輸入：Apr 26 12:20:02 localhost systemd[1]: Starting system activity accounting tool

filter{
    dissect{
        mapping => {
            "message" => "%{datetime_flie} %{+datetime_flie} %{+datetime_flie} %{local_host} %{sys_file}[%{pid}]: %{message_flle}"
        }
    }
}

Mutate插件：是使用最頻繁的插件，可以對字段進行各種操作(比如重命名、刪除、替換、更新等)
convert：類型轉換
gsub字符串替換
split、join、merge字符串切割、數組合并為字符串、數組合并為數組
rename字段重命名
update、replace字段內容更新或替換。它們都可以更新字段的內容，區別在于update只在字段存在時生效，而replace在字段不存在時會執行新增字段的操作
remove_field刪除字段
例子：對數據進行處理分割、添加字段和刪除字段
控制臺輸入：hello Weiking 20200810 man

filter{
    mutate{
        split => ["message", " "]
        add_field => {
            "field_name" => "%{[message][0]}"
            "name" => "%{[message][1]}"
            "date" => "%{[message][2]}"
            "sex" => "%{[message][3]}"
        }
        remove_field => "sex"
    }
}

Json插件：將指定字段的json格式的字符串進行解析，如果不指定target，會把解析出來的json數據直接放到根級別，可以通過target將解析后的json數據放到指定的字段中
例子：將json格式的字符串解析放在trans_all字段中
控制臺輸入：{"trans_jnls_no":"0","card_no":"623061571015020279","terminal_no":"zhanglq","atm_trans_types_type":"","customer_no":"106913609","cash_mark":"212"}

filter{
    json{
        source => "message"
        target => "trans_all"
        remove_field => "message"
    }
}

Geoip插件：基于Geoip庫解析IP數據獲取地域信息
GeoIP庫：可以根據IP地址提供對應的地域信息(包括國別，省市，經緯度等)
例子：解析IP地址獲取地域信息
控制臺輸入：124.160.75.205

filter{
    geoip{
        source => "message"
        remove_field => "message"
    }
}

output

標準輸出插件：多用于調試,從控制臺輸出

output {
    stdout {
        codec => rubydebug
    }
}

elasticsearch插件：將數據輸入到elaticsearch中

output {
    elasticsearch {
        #es的ip和端口
        hosts => ["http://192.168.57.100:9200"] 
        #集群：hosts => ["10.0.xx.xx:9200", "10.0.xx.xx:9200", "10.0.xx.xx:9200"]        
        #ES索引名稱（自己定義的）
        index => "mdl_001_202008"
        #文檔類型
        document_type => "MdL_001"
        #設置數據的id為數據庫中的字段
        document_id => "%{risk_event_no}-%{rule_id}"
    }
}

logstash配置實例

logstash的配置的文件：input和output都可以配置多個不同的插件，filter可以針對input里面的每個數據源做不一樣的過濾，通過各自定義的type來匹配
配置實例：

input{
    #從控制臺輸入數據
    stdin{ 
        type => "stdin"
    }  
    #讀取文件中的數據
    file {
        # 要導入的文件的位置，
        path => "/opt/data/data.json"            
        # 從文件開始的位置開始讀,end表示從結尾開始讀
        start_position => "beginning"
        # 多久之內沒修改過的文件不讀取，0為無限制，單位為秒
        ignore_older => 0  
        # 記錄文件上次讀取位置，輸出到null表示每次都從文件首行開始解析
        sincedb_path => "/dev/null"
        # type字段，可表明導入的日志類型
        type => "file"
    }
}
filter{
        if[type] == "stdin"{
            mutate{
                rename => ["message","timestamp"]
            }
        }
        if[type] == "file"{
            json{
                source => "message"
            }
        }
}
output {
        if[type] == "file"{
          elasticsearch {
                #es的ip和端口
                hosts => ["http://192.168.57.100:9200"] 
                #集群：hosts => ["10.0.xx.xx:9200", "10.0.xx.xx:9200", "10.0.xx.xx:9200"]        
                #ES索引名稱（自己定義的）
                index => "mdl_001_20200812"
                #文檔類型
                document_type => "MdL_012"
                #設置數據的id為數據庫中的字段
                document_id => "%{risk_event_no}"
            }

        }
        if[type] == "stdin"{
            stdout{
                codec => rubydebug
            }
        }
}

總結

以上是生活随笔為你收集整理的Logstash的简单介绍的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。