Elasticsearch 是一個開源的、基于 Lucene 的分布式搜索和分析引擎，設(shè)計用于云計算環(huán)境中，能夠?qū)崿F(xiàn)實時的、可擴(kuò)展的搜索、分析和探索全文和結(jié)構(gòu)化數(shù)據(jù)。它具有高度的可擴(kuò)展性，可以在短時間內(nèi)搜索和分析大量數(shù)據(jù)。

Elasticsearch 不僅僅是一個全文搜索引擎，它還提供了分布式的多用戶能力，實時的分析，以及對復(fù)雜搜索語句的處理能力，使其在眾多場景下，如企業(yè)搜索，日志和事件數(shù)據(jù)分析等，都有廣泛的應(yīng)用。

本文將介紹 Elastic Stack 組件 Logstash 的介紹、原理、安裝與簡單使用。

1、Logstash介紹與原理

1.1、Logstash簡介

Logstash 是一個開源的數(shù)據(jù)收集引擎，它具有實時管道功能，可以用來統(tǒng)一處理來自不同源的數(shù)據(jù)，并將其發(fā)送到你選擇的目標(biāo)。Logstash 支持多種類型的輸入數(shù)據(jù)，包括日志文件、系統(tǒng)消息隊列、數(shù)據(jù)庫等，可以對數(shù)據(jù)進(jìn)行各種轉(zhuǎn)換和處理，然后將數(shù)據(jù)發(fā)送到各種目標(biāo)，如 Elasticsearch、Kafka、郵件通知等。

Logstash 的主要特點(diǎn)包括：

1. 多輸入源：Logstash 支持多種類型的輸入數(shù)據(jù)，包括日志文件、系統(tǒng)消息隊列、數(shù)據(jù)庫等。

2. 數(shù)據(jù)處理：Logstash 可以對數(shù)據(jù)進(jìn)行各種轉(zhuǎn)換和處理，如過濾、解析、格式化等。

3. 多輸出目標(biāo)：Logstash 可以將數(shù)據(jù)發(fā)送到各種目標(biāo)，如 Elasticsearch、Kafka、郵件通知等。

4. 插件機(jī)制：Logstash 提供了豐富的插件，可以方便地擴(kuò)展其功能。

5. 與 Elasticsearch 和 Kibana 的集成：Logstash 是 Elastic Stack（前稱 ELK Stack）的一部分，與 Elasticsearch 和 Kibana 有很好的集成，可以方便地進(jìn)行數(shù)據(jù)搜索、存儲和可視化。

1.2、Logstash工作原理

Logstash 的工作原理可以分為三個主要步驟：輸入（Input）、過濾（Filter）和輸出（Output）。

1. 輸入（Input）：Logstash 支持多種類型的輸入數(shù)據(jù)，包括日志文件、系統(tǒng)消息隊列、數(shù)據(jù)庫等。在配置文件中，你可以指定一個或多個輸入源。
2. 過濾（Filter）：輸入數(shù)據(jù)被收集后，Logstash 可以對數(shù)據(jù)進(jìn)行各種轉(zhuǎn)換和處理。例如，你可以使用 grok 插件來解析非結(jié)構(gòu)化的日志數(shù)據(jù)，將其轉(zhuǎn)換為結(jié)構(gòu)化的數(shù)據(jù)。你也可以使用 mutate 插件來修改數(shù)據(jù)，如添加新的字段、刪除字段、更改字段的值等。
3. 輸出（Output）：處理后的數(shù)據(jù)可以被發(fā)送到一個或多個目標(biāo)。Logstash 支持多種類型的輸出目標(biāo)，包括 Elasticsearch、Kafka、郵件通知等。

這三個步驟是在 Logstash 的事件處理管道中順序執(zhí)行的。每個事件（例如，一行日志數(shù)據(jù)）都會經(jīng)過輸入、過濾和輸出這三個步驟。在過濾階段，如果一個事件被過濾器丟棄，那么它將不會被發(fā)送到輸出目標(biāo)。

以上就是 Logstash 的基本工作原理。需要注意的是，Logstash 的配置是非常靈活的，你可以根據(jù)實際需求來選擇合適的輸入源、過濾器和輸出目標(biāo)。

1.3、Logstash執(zhí)行模型

是的，你的理解是正確的。Logstash 的執(zhí)行模型主要包括以下幾個步驟：

1. 每個 Input 啟動一個線程：Logstash 會為每個輸入插件啟動一個線程，這些線程并行運(yùn)行，從各自的數(shù)據(jù)源獲取數(shù)據(jù)。

2. 數(shù)據(jù)寫入隊列：輸入插件獲取的數(shù)據(jù)會被寫入一個隊列。默認(rèn)情況下，這是一個存儲在內(nèi)存中的有界隊列，如果 Logstash 意外停止，隊列中的數(shù)據(jù)會丟失。為了防止數(shù)據(jù)丟失，Logstash 提供了兩個特性：

   • Persistent Queues：這個特性會將隊列存儲在磁盤上，即使 Logstash 意外停止，隊列中的數(shù)據(jù)也不會丟失。

   • Dead Letter Queues：這個特性會保存無法處理的事件。需要注意的是，這個特性只支持 Elasticsearch 作為輸出源。

3. 多個 Pipeline Worker 處理數(shù)據(jù)：Logstash 會啟動多個 Pipeline Worker，每個 Worker 會從隊列中取出一批數(shù)據(jù)，然后執(zhí)行過濾器和輸出插件。Worker 的數(shù)量和每次處理的數(shù)據(jù)量可以在配置文件中設(shè)置。

這種模型使得 Logstash 能夠高效地處理大量的數(shù)據(jù)，并且可以通過調(diào)整配置來優(yōu)化性能。

1.4、Logstash下載安裝

Elastic 公司的官方下載頁面的鏈接。在這個頁面上，你可以下載 Elastic Stack 的各個組件，包括 Elasticsearch、Kibana、Logstash、Beats 等。這個頁面提供了各個組件的最新版本下載鏈接，以及歷史版本的下載鏈接：Past Releases of Elastic Stack Software | Elastic

在這里，我們將選擇 Logstash，并確保所選的 Logstash 版本與我們正在使用的 Elasticsearch 版本一致：

選擇后選擇「Download」開始下載，并在下載成功后解壓到指定位置即可。

2、Logstash配置說明

2.1、Logstash配置介紹

Logstash 的配置主要分為兩部分：Pipeline 配置文件和 Settings 配置文件。

1. Pipeline 配置文件：這是 Logstash 的核心配置，用于定義數(shù)據(jù)處理的流程，包括輸入（input）、過濾（filter）和輸出（output）三個部分。每個部分都可以使用多種插件來完成特定的任務(wù)。例如，輸入部分可以使用 file 插件從文件中讀取數(shù)據(jù)，過濾部分可以使用 grok 插件解析日志，輸出部分可以使用 elasticsearch 插件將數(shù)據(jù)發(fā)送到 Elasticsearch。
2. Settings 配置文件：這是 Logstash 的全局配置，通常在 logstash.yml 文件中設(shè)置。這些配置包括 Logstash 實例的名稱、數(shù)據(jù)存儲路徑、配置文件路徑、自動重載配置、工作線程數(shù)量等。

這兩部分的配置都是以 YAML 格式編寫的，可以使用文本編輯器進(jìn)行編輯。在 Logstash 啟動時，它會首先讀取 Settings 配置文件，然后加載并執(zhí)行 Pipeline 配置文件。

2.2、Pipeline配置文件-輸入

在 Logstash 的 Pipeline 配置文件中，輸入（input）部分定義了數(shù)據(jù)的來源。Logstash 提供了多種輸入插件，可以從各種數(shù)據(jù)源讀取數(shù)據(jù)。

以下是一些常用的輸入插件：

file：從文件中讀取數(shù)據(jù)。常用的配置項包括 path（文件路徑）和 start_position（開始讀取的位置）。

input {
  file {
    path => "/path/to/your/logfile"
    start_position => "beginning"
  }
}

beats：從 Beats 客戶端（如 Filebeat、Metricbeat 等）接收數(shù)據(jù)。常用的配置項包括 port（監(jiān)聽的端口號）。

input {
  beats {
    port => 5044
  }
}

http：通過 HTTP 請求接收數(shù)據(jù)。常用的配置項包括 port（監(jiān)聽的端口號）。

input {
  http {
    port => 8080
  }
}

jdbc：從數(shù)據(jù)庫中讀取數(shù)據(jù)。常用的配置項包括 jdbc_driver_library（JDBC 驅(qū)動的路徑）、jdbc_driver_class（JDBC 驅(qū)動的類名）、jdbc_connection_string（數(shù)據(jù)庫連接字符串）、jdbc_user（數(shù)據(jù)庫用戶名）和 jdbc_password（數(shù)據(jù)庫密碼）。

input {
  jdbc {
    jdbc_driver_library => "/path/to/your/jdbc/driver"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://localhost:3306/yourdatabase"
    jdbc_user => "yourusername"
    jdbc_password => "yourpassword"
  }
}

kafka：在這個配置中，bootstrap_servers 參數(shù)指定了 Kafka 服務(wù)器的地址和端口，topics 參數(shù)指定了你想從哪個主題讀取數(shù)據(jù)。

input {
  kafka {
    bootstrap_servers => "localhost:9092"
    topics => ["your_topic"]
  }
}

kafka 輸入插件還有許多其他的配置項，你可以根據(jù)實際需求進(jìn)行設(shè)置。例如，你可以設(shè)置 group_id 參數(shù)來指定消費(fèi)者組，設(shè)置 auto_offset_reset 參數(shù)來指定在沒有初始偏移量或當(dāng)前偏移量不存在時該如何定位消費(fèi)位置等。

具體的配置項和可能的值，你可以在 Logstash 的官方文檔中找到。

以上就是一些常用的輸入插件及其配置。你可以根據(jù)實際需求選擇合適的插件和配置。需要注意的是，你可以在一個配置文件中定義多個輸入，Logstash 會并行處理所有的輸入。

2.3、Pipeline配置文件-過濾

在 Logstash 的 Pipeline 配置文件中，過濾（filter）部分定義了數(shù)據(jù)處理的規(guī)則。過濾器插件可以對數(shù)據(jù)進(jìn)行各種操作，如解析、轉(zhuǎn)換、添加和刪除字段等。

以下是一些常用的過濾插件及其操作：

grok：grok 過濾器用于解析非結(jié)構(gòu)化的日志數(shù)據(jù)，將其轉(zhuǎn)換為結(jié)構(gòu)化的數(shù)據(jù)。它使用模式匹配的方式來解析文本，每個模式是一個名字和正則表達(dá)式的組合。例如：

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
}

在這個配置中，grok 過濾器會嘗試將 message 字段的內(nèi)容匹配為 COMBINEDAPACHELOG 模式，這是一個預(yù)定義的模式，用于解析 Apache 日志。

mutate：mutate 過濾器用于修改事件數(shù)據(jù)，如添加新的字段、刪除字段、更改字段的值等。例如：

filter {
  mutate {
    add_field => { "new_field" => "new_value" }
  }
}

在這個配置中，mutate 過濾器會向每個事件添加一個名為 new_field 的新字段，字段的值為 new_value。

date：date 過濾器用于解析日期和時間信息，將其轉(zhuǎn)換為 Logstash 的 @timestamp 字段。例如：

filter {
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

在這個配置中，date 過濾器會嘗試將 timestamp 字段的內(nèi)容匹配為指定的日期和時間格式。

以上就是一些常用的過濾插件及其操作。你可以根據(jù)實際需求選擇合適的插件和配置。需要注意的是，你可以在一個配置文件中定義多個過濾器，Logstash 會按照配置文件中的順序依次執(zhí)行這些過濾器。

2.4、Pipeline配置文件-輸出

在 Logstash 的 Pipeline 配置文件中，輸出（output）部分定義了處理后的數(shù)據(jù)應(yīng)該發(fā)送到哪里。Logstash 提供了多種輸出插件，可以將數(shù)據(jù)發(fā)送到各種目標(biāo)。

以下是一些常用的輸出插件：

elasticsearch：將數(shù)據(jù)發(fā)送到 Elasticsearch。常用的配置項包括 hosts（Elasticsearch 服務(wù)器的地址和端口）和 index（索引名稱）。

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "your_index"
  }
}

file：將數(shù)據(jù)寫入到文件。常用的配置項包括 path（文件路徑）。

output {
  file {
    path => "/path/to/your/file"
  }
}

stdout：將數(shù)據(jù)輸出到標(biāo)準(zhǔn)輸出。常用的配置項包括 codec（編碼格式），常用的值有 rubydebug（以 Ruby 的調(diào)試格式輸出）。

output {
  stdout {
    codec => rubydebug
  }
}

kafka：將數(shù)據(jù)發(fā)送到 Kafka。常用的配置項包括 bootstrap_servers（Kafka 服務(wù)器的地址和端口）和 topic_id（主題名稱）。

output {
  kafka {
    bootstrap_servers => "localhost:9092"
    topic_id => "your_topic"
  }
}

以上就是一些常用的輸出插件及其配置。你可以根據(jù)實際需求選擇合適的插件和配置。需要注意的是，你可以在一個配置文件中定義多個輸出，Logstash 會將每個事件發(fā)送到所有的輸出。

2.4、Settings配置文件

Logstash 的 Settings 配置文件通常是 logstash.yml，這是 Logstash 的全局配置文件，用于設(shè)置 Logstash 運(yùn)行的一些基本參數(shù)。

以下是一些常見的配置項：

1. node.name：設(shè)置 Logstash 實例的名稱，默認(rèn)值為當(dāng)前主機(jī)的主機(jī)名

   node.name: test
   

2. path.data：設(shè)置 Logstash 存儲持久化數(shù)據(jù)的路徑，默認(rèn)值為 Logstash 安裝目錄下的 data 文件夾

   path.data: /var/lib/logstash
   

3. path.config：設(shè)置 Pipeline 配置文件的路徑。

   path.config: /etc/logstash/conf.d/*.conf
   

4. config.reload.automatic：如果設(shè)置為 true，Logstash 會自動檢測 Pipeline 配置文件的更改，并重新加載配置。

   config.reload.automatic: true
   

5. pipeline.workers：設(shè)置處理事件的工作線程數(shù)量，通常設(shè)置為機(jī)器的 CPU 核心數(shù)。

   pipeline.workers: 2
   

6. pipeline.batch.size：設(shè)置每個批處理的事件數(shù)量，增大這個值可以提高吞吐量，但也會增加處理延遲。

   pipeline.batch.size: 125
   

7. pipeline.batch.delay：設(shè)置兩個批處理之間的最大等待時間（以毫秒為單位）。

   pipeline.batch.delay: 50
   

以上就是一些常見的 Logstash Settings 配置項，你可以根據(jù)實際需求來修改這些配置。具體的配置項和可能的值，你可以在 Logstash 的官方文檔中找到。

3、Logstash使用示例

3.1、Logstash Hello world

首先我們進(jìn)行一個非?；A(chǔ)的 Logstash 使用示例。在這個示例中，Logstash 使用標(biāo)準(zhǔn)輸入作為輸入源，標(biāo)準(zhǔn)輸出作為輸出目標(biāo)，且不指定任何過濾器。

1. 在命令行中切換到 Logstash 的根目錄，然后執(zhí)行以下命令來啟動 Logstash：

cd logstash-8.10.2
bin/logstash -e 'input { stdin { } } output { stdout {} }'

在這個命令中，-e 參數(shù)用于指定 Pipeline 配置，input { stdin { } } 表示使用標(biāo)準(zhǔn)輸入作為輸入源，output { stdout {} } 表示使用標(biāo)準(zhǔn)輸出作為輸出目標(biāo)。

2. Logstash 啟動成功后，你可以在控制臺輸入一些文本，如 “hello world”，然后 Logstash 會將這些文本作為事件數(shù)據(jù)處理。

3. Logstash 會自動為每個事件添加一些字段，如 @version、host 和 @timestamp，然后將處理后的事件輸出到標(biāo)準(zhǔn)輸出。

例如，你在控制臺輸入 “hello world” 后，可能會看到如下的輸出：

{
    "@version": "1",
    "host": "localhost",
    "@timestamp": "2018-09-18T12:39:38.514Z",
    "message": "hello world"
}

在這個示例中，Logstash 僅僅是從標(biāo)準(zhǔn)輸入獲取數(shù)據(jù)，添加一些簡單的字段，然后將數(shù)據(jù)輸出到標(biāo)準(zhǔn)輸出。這是最基礎(chǔ)的使用方式，實際上 Logstash 還可以做很多復(fù)雜的數(shù)據(jù)處理和轉(zhuǎn)換。

3.2、日志格式處理

我們可以看到雖然上面示例使用標(biāo)準(zhǔn)輸入作為輸入源，并將數(shù)據(jù)輸出到標(biāo)準(zhǔn)輸出，但是日志內(nèi)容作為一個整體被存放在 message 字段中，這樣對后續(xù)存儲及查詢都極為不便。可以為該 pipeline 指定一個 grok filter 來對日志格式進(jìn)行處理。

1. 在 first-pipeline.conf 中增加 filter 配置如下

input { stdin { } }
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}
output {
   stdout { codec => rubydebug }
}

其中 codec => rubydebug 用于美化輸出

2. 驗證配置（注意指定配置文件的路徑）：

./bin/logstash -f first-pipeline.conf --config.test_and_exit 

3. 啟動命令:

./bin/logstash -f first-pipeline.conf --config.reload.automatic 

其中 --config.reload.automatic 選項啟用動態(tài)重載配置功能

4. 預(yù)期結(jié)果:

我們的配置中使用了 grok 過濾器來解析 COMBINEDAPACHELOG 格式的 Apache 日志。以下是一個符合這種格式的日志示例：

127.0.0.1 - - [28/Sep/2021:10:00:00 +0800] "GET /test.html HTTP/1.1" 200 2326 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"

這條日志記錄了一個 HTTP 請求的信息，包括客戶端 IP 地址、請求時間、請求方法和 URL、HTTP 版本、響應(yīng)狀態(tài)碼、響應(yīng)體的字節(jié)數(shù)、Referer 和 User-Agent 等。

我們可以將這個日志作為輸入，然后 Logstash 會使用我們的配置來處理這個日志。處理后的結(jié)果會被輸出到標(biāo)準(zhǔn)輸出，格式為 Ruby 的調(diào)試格式。

3.3、將數(shù)據(jù)導(dǎo)入Elasticsearch

Logstash 作為 Elastic stack 的重要組成部分，其最常用的功能是將數(shù)據(jù)導(dǎo)入到 Elasticssearch 中。將 Logstash 中的數(shù)據(jù)導(dǎo)入到 Elasticsearch 中操作也非常的方便，只需要在 pipeline 配置文件中增加 Elasticsearch 的 output 即可。

1. 首先要有一個已經(jīng)部署好的 Logstash
2. 在 first-pipeline.conf 中增加 Elasticsearch 的配置，如下

input { stdin { } }
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}
output {
    elasticsearch {
        hosts => [ "localhost:9200" ]
        topic_id => "logstash"
    }
}

3. 啟動命令:

./bin/logstash -f first-pipeline.conf --config.reload.automatic 

其中 --config.reload.automatic 選項啟用動態(tài)重載配置功能

4. 預(yù)期結(jié)果:

我們的配置中使用了 grok 過濾器來解析 COMBINEDAPACHELOG 格式的 Apache 日志。以下是一個符合這種格式的日志示例：

"127.0.0.1 - - [28/Sep/2021:10:00:00 +0800] "GET /test.html HTTP/1.1" 200 2326 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"

這條日志記錄了一個 HTTP 請求的信息，包括客戶端 IP 地址、請求時間、請求方法和 URL、HTTP 版本、響應(yīng)狀態(tài)碼、響應(yīng)體的字節(jié)數(shù)、Referer 和 User-Agent 等。

我們可以將這個日志作為輸入，然后 Logstash 會使用我們的配置來處理這個日志。處理后的結(jié)果會被輸出到標(biāo)準(zhǔn)輸出，格式為 Ruby 的調(diào)試格式。

查詢 Elasticsearch 確認(rèn)數(shù)據(jù)是否正常上傳：

curl -XGET 'http://localhost:9200/logstash/_search?pretty&q=response=200'

也可以使用 Kibana 查看：

文章來源地址http://www.zghlxwxcb.cn/news/detail-712924.html

到了這里，關(guān)于Elasticsearch系列組件：Logstash強(qiáng)大的日志管理和數(shù)據(jù)分析工具的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！