kafka 大数据日志平台服务搭建设计

起因：在实际项目开发过程中，需要使用RabbitMQ来实现消息队列的功能，在运用过之后，也去学一学kafka，了解一下他们之间的差别，吃一吃架构方面的相关内容，提升自己。

---

1. 大数据日志平台服务搭建设计

收集日志要注意的是

• 日志信息量非常大，所以一定要有一个可以堆积的地方

• 日志会出现暴增的情况

• 日志文件-->kafka(分布式)-->Logstash(收集过滤)-->ElasticSearch(聚合计算)-->Kibana(报表展示)

2. 使用filebeat将日志收集进kafka

logstash：使用java开发的比较消耗资源

filebeat：使用go语言开发的的空间，比logstash更轻量级，占用资源更少，一般在生产环境都是使用filebeat来进行日志收集的

# 下载安装
wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.5.2-linux-x86_64.tar.gz
# 解压
tar -zxvf filebeat-7.5.2-linux-x86_64.tar.gz
# 修改配置文件
vi /usr/local/filebeat/filebeat.yml

filebeat.yml的配置信息

filebeat.inputs:
# 我的这个输入项是干什么的,自己命个名标记下
- input_type: log

 paths:
 - /usr/local/logsfile/error.log
 document_type: "error-log"    # 写入ES时的_type值
 multiline:
 pattern: '^\['              # 指定匹配的表达式,匹配以[开头的字符串
 negate: true                # 是否匹配到,必须要匹配到
 match: after                # 没有匹配到的合并到上一行的末尾
 max_lines: 2000             # 未匹配到最大的行数
 timeout: 2s                 # 如果在规定时间没有新的日志事件就不等待后面的直接output
 fields: 
 logbiz: order-error         # 业务命名
 logtopic: order-error-log   # kafka的topic
 evn: dev                    # 定义一个环境名
​
- input_type: log

 paths:
 - /usr/local/logsfile/info.log
 document_type: "info-log"
 multiline:
 pattern: '^\['
 negate: true
 match: after
 max_lines: 2000
 timeout: 2s
 fields: 
 logbiz: order-info
 logtopic: order-info-log
 evn: dev
​
​
output.kafka: 
 enabled: true
 hosts: ["192.168.0.177:9092","192.168.0.178:9092","192.168.0.179:9092"]
 topic: '%{[fields.logtopic]}'
 partition.hash: 
 reachable_only: true
 compression: gzip
 max_message_bytes: 1000000
 required_acks: 1
logging.to_files: true

启动filebeat

# 启动前做一下验证,看一下配置文件是否正确
./filebeat test config
# 在启动之前一定要将filebeat导入的topic创建好
# 后台启动filebeat
./filebeat &

3. 通过logstash提取kafka数据进入ES

# 0.解压安装
# 1.在根目录创建sync的
# 2.在sync文件里创建logstash-log-sync.conf,配置内容如下
input {
 kafka {
 bootstrap_servers => "192.168.0.177:9092,192.168.0.178:9092,192.168.0.179:9092"
 topics_pattern  => "order-.*"
 consumer_threads => 5
 decorate_events => true
 codec => "json"
 auto_offset_reset => "latest"
 group_id => "logstash1"
 }

}
output {
 elasticsearch {
 hosts => ["192.168.0.175:9200"]
 index => "kafka-%{+YYYY.MM.dd}"
 }
 stdout{
 codec => rubydebug     ##输出到屏幕上
 }
}
# 在bin目录下启动执行
./logstash -f /usr/local/logstash/sync/logstash-log-sync.conf

---

不要以为每天把功能完成了就行了，这种思想是要不得的，互勉~！