背景介紹

• 公司業務系統做優化改造，同時為了能夠實現全鏈路監控，需收集所有業務系統之間的調用日志。
• 數據情況:每天20億+
• 機器成本:3臺kafka集群，2臺logstash采集機器
• 技術:Java,MQ,MLSQL,Logstash

[[340286]]

下圖為最終結果圖

采集流程 

流程分解

流程一：MLSQL 消費MQ

原始日志產生側通過protobuf進行序列化推送至mq,然后通過MLSQL進行反序列化并進行簡單的etl處理后，再推送至MQ

流程二：通過Logstash進行消費MQ

通過logstash消費經過MLSQL處理后的數據，并在這里通過ruby進行再次的加工處理，最后寫入es和hdfs

注意：這里一部分流程推送到es是業務側使用，而另一部分寫入hdfs是提供給數倉使用

流程三: 數倉建模

這里通過數倉建模，將最后的指標結果推送至es提供給業務側使用

注意：本篇主要是借鑒這個需求講解Logstash在實際場景中的使用以及優化，其他兩個部分流程不做詳細講解

為什么這樣設計?

原因一：

首先這個需求屬于日志采集的范疇，但Logstash本身不支持反序列化功能，需要自定義開發ruby插件來支持，但這樣一來開發成本較高，且不好維護，所以使用了MLSQL結合UDF的方式進行流式處理

原因二：

大家在對最后的輸出流程可能會有疑惑，為什么不直接通過MLSQL來寫入到hdfs和es呢，這里有兩點：

• MLSQL寫入hdfs會產生大量的小文件，需要單獨開發合并文件的功能
• 最后寫入es的數據是需要數倉結合其他業務數據進行建模的，而MLSQL在這點做的不太好，所以這里走的是離線處理的方式

說到這里，具體的場景需要結合公司的實際情況來進行決策的，有些同學或許會想為什么不用flume進行日志采集呢?那這里就不做過多的解釋了，白菜蘿卜各有所愛，適合自己的才是最好的!廢話不多說，接下來進入正題，結合該需求場景，如何使用較少的成本完成大數據量的采集呢?以及如何優化呢?

Logstash開發流程

1.確定日志格式

首先呢，一個日志文件里肯定是不止一種日志格式，也有可能是標準化的格式，這里需要跟日志產生側進行確認格式

2.調試grok

確定好日志格式后，編寫grok語法，然后進行調試，本人是通過kibana6自帶的grok debug進行調試。結合該需求背景，最后經過logstash采集的時候，其實已經通過MLSQL進行了處理，最后Logstash消費的是格式就是一個json字符串，所以不需要grok語法，但是這里還是簡單舉個例子來說明一下 

3.調試ruby

結合該需求，使用ruby進行一些清洗邏輯 

4.優化

這里優化的工作在整個需求開發周期的比例較大，因為數據量較大，且資源比較少，具體優化思路如下：

1.MLSQL優化

這部分的優化工作主要是在反序化這塊，剔除了一部分無用字段，以及提前過濾了一部分數據量，這里給出一部分注冊UDF的代碼 

2.Kafka端優化

因kafka集群是集團共用,所以kafka端的優化其實只涉及到消費端的優化。這里只調節了兩個參數

1. ：數據壓縮
2. ：消費者線程數

3.hdfs優化

logstash寫入hdfs的部分不用使用自帶的webhdfs插件，而是自定義的插件。

因自定義插件中涉及到文件鎖的問題，會通過比對前后兩次文件是否一致來進行文件最后的刷寫，所以這里只能通過減少文件的更新頻率來減少上下文的切換以及刷寫操作

4.ES優化

es部分的優化也只是涉及到寫優化，比如批量寫入、調大線程數、增加refresh間隔、禁止swapping交換內存、禁止refresh和replica操作，調大index buffer等操作。