環(huán)境

• win10系統(tǒng)
  • filebeat 7.10.0
• CentOS7 集群（虛擬機(jī)）
  • kafka 0.11

第一步 下載filebeat

下載地址

我下載的是圖中圈出的壓縮包版本，也有MSI版本供下載。

第二步 配置filebeat

官方教程

1. 將下載的壓縮包解壓到“C:\Program Files\Filebeat”。
2. 修改該目錄中的配置文件filebeat.yml，主要修改以下幾個(gè)地方：

• 添加filebeat要采集的源文件相關(guān)信息，主要在inputs中。

# ============================== Filebeat inputs ===============================

filebeat.inputs:

# Each - is an input. Most options can be set at the input level, so
# you can use different inputs for various configurations.
# Below are the input specific configurations.

- type: log

  # Change to true to enable this input configuration.
  enabled: true #這里一定要改成true才會(huì)生效

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    #- /var/log/*.log 這是linux中的路徑
    #- c:\programdata\elasticsearch\logs\*
    - e:\Lab\Logs\test.log #日志文件路徑

添加輸出信息（kafka），主要在output中。

# ---------------------------- kafka Output----------------------------
output.kafka:
  # initial brokers for reading cluster metadata hosts是string數(shù)組類型，支持多個(gè)kafka實(shí)例地址(此處不是配置zookeeper地址)
  hosts: ['XXXX:9092']

  # message topic selection + partitioning
  #發(fā)送到kafka的那個(gè)topic，我這里事先在fields中配置了一個(gè)kafka_topic字段，下邊有寫
  topic: '%{[fields.kafka_topic]}'
  version: 0.11.0.0 #kafka版本
  partition.round_robin: #向kafka分區(qū)發(fā)送的方式（這里是輪詢）
    reachable_only: false

  required_acks: 1 #kafka應(yīng)答方式
  compression: gzip
  max_message_bytes: 1000000

自己添加的字段，kafka的topic名稱

# ================================== General ===================================

# The name of the shipper that publishes the network data. It can be used to group
# all the transactions sent by a single shipper in the web interface.
#name:

# Optional fields that you can specify to add additional information to the
# output.
fields:
  kafka_topic: first #自己定義的字段 表示kafka中的topic 名稱為first

配置完以后保存即可。

第三步 開啟filebeat

此時(shí)部署在linux虛擬機(jī)上的kafka集群已啟動(dòng)，不再贅述。

1. 以管理員身份打開powershell，進(jìn)入filebeat安裝目錄'C:\Program Files\Filebeat',輸入以下命令，使用剛修改好的配置文件啟動(dòng)filebeat。

PS C:\Program Files\Filebeat> .\filebeat -e -c filebeat.yml

注意一定要以管理員身份打開。
　　如果powershell報(bào)”系統(tǒng)中禁止執(zhí)行腳本'，就輸入以下命令更改執(zhí)行策略：

PS C:\Program Files\Filebeat> set-ExecutionPolicy RemoteSigned

記得也要在管理員權(quán)限下才行。
　　另外一提，powershell的cmd的cd命令有點(diǎn)區(qū)別，當(dāng)目錄名字帶空格時(shí)，需要將目標(biāo)地址寫進(jìn)引號(hào)里，形成一個(gè)參數(shù)，否則powershell切換會(huì)提示找不到目標(biāo)位置。

2. 如果問(wèn)題都解決完畢，filebeat正常開啟，此時(shí)只需要之前在配置文件的inputs中設(shè)置的log文件有新增行，filebeat就會(huì)把數(shù)據(jù)發(fā)送到Kafka中。
   　　在kafka中事先開啟一個(gè)kafka的consumer，方便看到filebeat傳輸?shù)臄?shù)據(jù)。

[root@centos7_1 kafka]# bin/kafka-console-consumer.sh --bootstrap-server CentOS7_1:9092 --topic first

3. 此時(shí)在”e:\Lab\Logs\test.log“中隨便輸入一些數(shù)據(jù)并保存，kafka的消費(fèi)者就會(huì)打印相應(yīng)數(shù)據(jù)，數(shù)據(jù)是json格式的：

{
    '@timestamp': '2020-12-01T07:49:19.035Z',
    '@metadata': {
        'beat': 'filebeat',
        'type': '_doc',
        'version': '7.10.0'
    },
    'fields': {
        'kafka_topic': 'first'
    },
    'host': {
        'name': xxx,
        'mac': [
            xxx
        ],
        'hostname': xxx,
        'architecture': 'x86_64',
        'os': {
            'build': '18363.1198',
            'platform': 'windows',
            'version': '10.0',
            'family': 'windows',
            'name': 'Windows 10 Home China',
            'kernel': '10.0.18362.1198 (WinBuild.160101.0800)'
        },
        'id': xx,
        'ip': [
            xxx
        ]
    },
    'agent': {
        'ephemeral_id': '8a09326d-8a07-41cc-aba4-6735eec57c63',
        'id': '287d0b6a-2dc7-427e-af59-2acea9a2b89e',
        'name': 'LAPTOP-TP5O2I6E',
        'type': 'filebeat',
        'version': '7.10.0',
        'hostname': xxx
    },
    'ecs': {
        'version': '1.6.0'
    },
    'log': {
        'file': {
            'path': 'e:\\Lab\\Logs\\test.log'
        },
        'offset': 14
    },
    'message': '1231323231132',
    'input': {
        'type': 'log'
    }
}

message中便是我在日志文件中輸入的內(nèi)容，可以看到已經(jīng)成功發(fā)了過(guò)來(lái)。
但是，我就隨便打了一行，filebeat發(fā)過(guò)來(lái)變成這么多，有很多我不需要的數(shù)據(jù)，那么是否可以去掉哪些不需要的字段呢？當(dāng)然可以！

去除filebeat發(fā)送數(shù)據(jù)中不需要的字段

重新回到filebeat.yml文件，修改processors，在這里我們可以對(duì)filebeat發(fā)送的數(shù)據(jù)格式做一些處理。
官方介紹

# ================================= Processors =================================
processors:
  # 去掉這些自帶的處理器
  # - add_host_metadata:
  #     when.not.contains.tags: forwarded
  # - add_cloud_metadata: ~
  # - add_docker_metadata: ~
  # - add_kubernetes_metadata: ~
  # 使用drop_fields這個(gè)處理器去掉不想要的字段
    - drop_fields:
      # when: #可以設(shè)置條件，這里就不設(shè)置了
      #   condition
      # 要去除的字段
      fields: ['log','host','input','agent','ecs']
      ignore_missing: false #缺少字段是否報(bào)錯(cuò)

修改完成后再次運(yùn)行filebeat，并在log文件中輸入數(shù)據(jù)并保存，在kafka的消費(fèi)者可以看到這次收到的數(shù)據(jù)就清爽了許多：

{
   '@timestamp': '2020-12-01T09:11:04.239Z',
   '@metadata': {
       'beat': 'filebeat',
       'type': '_doc',
       'version': '7.10.0'
   },
   'message': '12312321',
   'fields': {
       'kafka_topic': 'first'
   }
}

注意:@timestamp和@metadata這兩個(gè)字段是去不掉的。