以下是我的 filebeats 配置文件：
-----------------
cat /etc/filebeat/filebeat.yml
-----------------
filebeat:
spool_size: 1024 # 最大可以攒够 1024 条数据一起发送出去
idle_timeout: "5s" # 否则每 5 秒钟也得发送一次
registry_file: ".filebeat" # 文件读取位置记录文件，会放在当前工作
目录下。所以如果你换一个工作目录执行 filebeat 会导致重复传输！
config_dir: "/etc/filebeat/conf"

output.kafka:
hosts: ["x.x.x.x:9091"]
topic: '%{[logstash_type]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
-----------------
cat /etc/filebeat/conf/pubservice.yml
-----------------
filebeat.prospectors:
-
tail_files: true
fields:
logstash_type: "pubservice"
paths:
- /var/log/pubservice.log
fields_under_root: true
json:
message_key: message
overwrite_keys: true
keys_under_root: true
-----------------
-----------------
我这边是走 filebeats -> kafka -> logstash -> elasticsearch 的数据流走向

logstash 根据『 logstash_type 』 字段做判断，塞进不同的 index 里面。

2: 没明白撒意思，手动建立 mapping 看需求，我是倾向于建的，这边不建立 mapping 的情况下，偶有数据类型不一致会丢

3: filebeats 只是抽数据的吧，建立 mapping 是在 ES 这里做的，你可以通过 es 的 template 接口去实现指定的 index 匹配一套 mapping.

4: 看上面的配置文件，grok 这个帮不了你，我也不熟

R7000 或者同级别封顶，高了真没撒意思。