**背景** 日志系統(tǒng)接入的日志種類多、格式復雜多樣，主流的有以下幾種日志： - Filebeat采集到的文本日志，格式多樣 - Winbeat采集到的操作系統(tǒng)日志 - 設(shè)備上報到Logstash的syslog日志 - 接入到Kafka的業(yè)務(wù)日志 以上通過各種渠道接入的日志，存在2個主要的問題： - 格式不統(tǒng)一、不規(guī)范、標準化不夠 - 如何從各類日志中提取出用戶關(guān)心的指標，挖掘更多的業(yè)務(wù)價值 為了解決上面2個問題，我們基于Flink和Drools規(guī)則引擎做了實時的日志處理服務(wù)。 **系統(tǒng)架構(gòu)** 架構(gòu)比較簡單，架構(gòu)圖如下： 各類日志都是通過Kafka匯總，做日志中轉(zhuǎn)。 Flink消費Kafka的數(shù)據(jù)，同時通過API調(diào)用拉取Drools規(guī)則引擎，對日志做解析處理后，將解析后的數(shù)據(jù)存儲到Elasticsearch中，用于日志的搜索和分析等業(yè)務(wù)。 為了監(jiān)控日志解析的實時狀態(tài)，F(xiàn)link會將日志處理的統(tǒng)計數(shù)據(jù)，如每分鐘處理的日志量，每種日志從各個機器IP來的日志量寫到Redis中，用于監(jiān)控統(tǒng)計。 **模塊介紹** 系統(tǒng)項目命名為Eagle。 eagle-api：基于Spring Boot，作為Drools規(guī)則引擎的寫入和讀取API服務(wù)。 eagle-common：通用類模塊。 eagle-log：基于Flink的日志處理服務(wù)。 重點講一下eagle-log： **對接kafka、ES和Redis** 對接Kafka和ES都比較簡單，用的官方的connector（flink-connector-kafka-0.10和flink-connector-elasticsearch6），詳見代碼。 對接Redis，最開始用的是org.apache.bahir提供的redis connector，后來發(fā)現(xiàn)靈活度不夠，就使用了Jedis。 在將統(tǒng)計數(shù)據(jù)寫入redis的時候，最開始用的keyby分組后緩存了分組數(shù)據(jù)，在sink中做統(tǒng)計處理后寫入，參考代碼如下： ``` String name = "redis-agg-log"; DataStream>> keyedStream = dataSource.keyBy((KeySelector) log -> log.getIndex()) .timeWindow(Time.seconds(windowTime)).trigger(new CountTriggerWithTimeout<>(windowCount, TimeCharacteristic.ProcessingTime)) .process(new ProcessWindowFunction>, String, TimeWindow>() { @Override public void process(String s, Context context, Iterable iterable, Collector>> collector) { ArrayList logs = Lists.newArrayList(iterable); if (logs.size() > 0) { collector.collect(new Tuple2(s, logs)); } } }).setParallelism(redisSinkParallelism).name(name).uid(name); ``` 后來發(fā)現(xiàn)這樣做對內(nèi)存消耗比較大，其實不需要緩存整個分組的原始數(shù)據(jù)，只需要一個統(tǒng)計數(shù)據(jù)就OK了，優(yōu)化后： ``` String name = "redis-agg-log"; DataStream keyedStream = dataSource.keyBy((KeySelector) log -> log.getIndex()) .timeWindow(Time.seconds(windowTime)) .trigger(new CountTriggerWithTimeout<>(windowCount, TimeCharacteristic.ProcessingTime)) .aggregate(new LogStatAggregateFunction(), new LogStatWindowFunction()) .setParallelism(redisSinkParallelism).name(name).uid(name); ``` 這里使用了Flink的聚合函數(shù)和Accumulator，通過Flink的agg操作做統(tǒng)計，減輕了內(nèi)存消耗的壓力。 **使用Broadcast廣播Drools規(guī)則引擎** 1、Drools規(guī)則流通過broadcast map state廣播出去。 2、Kafka的數(shù)據(jù)流connect規(guī)則流處理日志。 ``` //廣播規(guī)則流 env.addSource(new RuleSourceFunction(ruleUrl)).name(ruleName).uid(ruleName).setParallelism(1) .broadcast(ruleStateDescriptor); //Kafka數(shù)據(jù)流 FlinkKafkaConsumer010 source = new FlinkKafkaConsumer010<>(kafkaTopic, new LogSchema(), properties); env.addSource(source).name(kafkaTopic).uid(kafkaTopic).setParallelism(kafkaParallelism); //數(shù)據(jù)流connect規(guī)則流處理日志 BroadcastConnectedStream connectedStreams = dataSource.connect(ruleSource); connectedStreams.process(new LogProcessFunction(ruleStateDescriptor, ruleBase)).setParallelism(processParallelism).name(name).uid(name); ``` 具體細節(jié)參考開源代碼。 **小結(jié)** 本系統(tǒng)提供了一個基于Flink的實時數(shù)據(jù)處理參考，對接了Kafka、Redis和Elasticsearch，通過可配置的Drools規(guī)則引擎，將數(shù)據(jù)處理邏輯配置化和動態(tài)化。 對于處理后的數(shù)據(jù)，也可以對接到其他sink，為其他各類業(yè)務(wù)平臺提供數(shù)據(jù)的解析、清洗和標準化服務(wù)。 > 【云棲號在線課堂】每天都有產(chǎn)品技術(shù)專家分享！ > 課程地址：https://yqh.aliyun.com/live > 立即加入社群，與專家面對面，及時了解課程最新動態(tài)！ > 【云棲號在線課堂 社群】https://c.tb.cn/F3.Z8gvnK 原文發(fā)布時間：2020-07-09 本文作者： aoxiang 本文來自：“”，了解相關(guān)信息可以關(guān)注“dockone”