ELK｜Integrate Kafka with Logstash and Beats to Elasticsearch

最後更新:2021-08-11 23:57:30 By ELK 達人 2417

kafka integrate beats and logstash

在服務企業的過程中，遇到一天 TB、PB 級的資料是常態，且可能在短時間內會有大量的 Log 資料寫入到 Elasticsearch 中，因此透過 Kafka 分擔工作量是很常使用的技巧之一，底下筆者將標準用法整理出來。

安裝

Filebeat

Logstash

Kafka (Azure EventHub)

安裝 Kafka

wget https://downloads.apache.org/kafka/2.8.0/kafka_2.13-2.8.0.tgz
tar -xzf kafka_2.13-2.8.0.tgz
cd kafka_2.13-2.8.0

啟動 Kafka 服務

bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

設定 Filebeat

Filebeat 做為一個 producer 的角色，指定一個 topic 推播工作消息。

需要調整的參數有：Log 路徑、Kafka Hostname、Kafka Topic、帳號、密碼。

    filebeat.inputs:
    - type: log
      enabled: true
      paths:
        - /your/log/path
    
    output.kafka:
      topic: elk-lab
      required_acks: 1
      version: '1.0.0'
      partition.round_robin:
        reachable_only: false
      hosts:
        - "your-kafka-domain:9092"
      #username: ""
      #password: ""
      #ssl.enabled: true
      compression: none
      
    logging:
      level: debug
      to_files: true
      to_syslog: false
      files:
        path: /opt/filebeat/log
        name: eventhub.log
        keepfiles: 7
        rotateeverybytes: 10485760 # 10 MB

設定 Logstash

Logstash 做為一個 consumer 的角色，可以處理多個 topics，並轉送至 elasticsearch。

    input {
        kafka {
          bootstrap_servers => "localhost:9092"
          topics => ["elk-lab"]
        }
      }
      
      output {
        elasticsearch {
          hosts => ["http://localhost:9200"]
          index => "logstash-kafka-%{+YYYY.MM.dd}"
        }
      }

可以進一步參考 grok filter plugin 處理 Log 資料