時(shí)日志分析平臺(tái)的Kappa架構(gòu)實(shí)戰(zhàn))
1. 項(xiàng)目整體設(shè)計(jì)與Kappa架構(gòu)選型背后的邏輯1.1 為什么是Kappa而不是Lambda先說結(jié)論如果你現(xiàn)在還要為一個(gè)新項(xiàng)目搭建實(shí)時(shí)日志分析平臺(tái)Lambda架構(gòu)大概率已經(jīng)不是最優(yōu)解了。Kappa架構(gòu)的核心思想非常樸素——把所有數(shù)據(jù)都當(dāng)作流來處理用一個(gè)引擎同時(shí)支撐實(shí)時(shí)計(jì)算和歷史數(shù)據(jù)重放不需要像Lambda那樣為批處理和流處理各維護(hù)一套代碼。Lambda架構(gòu)給人挖的坑我太有體會(huì)了。流批兩套代碼意味著兩套邏輯、兩套部署、兩套運(yùn)維最痛苦的是當(dāng)你要修一個(gè)bug或者加一個(gè)字段時(shí)要在兩個(gè)項(xiàng)目里分別改一遍然后還得對(duì)兩邊的計(jì)算結(jié)果做合并和校驗(yàn)。日志分析這個(gè)場(chǎng)景尤其尷尬日志數(shù)據(jù)本質(zhì)上就是一條條不斷產(chǎn)生的事件流你非要用批處理框架對(duì)一份靜態(tài)文件反復(fù)跑批屬于脫褲子放屁。Kappa的底氣來自Kafka的持久化和重放能力。Kafka可以保留全量日志數(shù)據(jù)通常按天或按容量設(shè)置retention當(dāng)業(yè)務(wù)方需要重新計(jì)算某個(gè)時(shí)間窗口的指標(biāo)時(shí)我們只要把Kafka的消費(fèi)位點(diǎn)重置到那個(gè)時(shí)間點(diǎn)之前再用Flink從那個(gè)位點(diǎn)重新消費(fèi)、重新計(jì)算結(jié)果寫到新的Elasticsearch索引里就行。整個(gè)過程不需要啟動(dòng)任何批處理任務(wù)也不需要寫一套MapReduce代碼。舉一個(gè)具體例子某天凌晨線上有一個(gè)支付接口的調(diào)用量異常飆升業(yè)務(wù)方想對(duì)比今天早高峰和上周同一天的數(shù)據(jù)。Lambda架構(gòu)的做法是臨時(shí)寫一個(gè)Hive SQL跑一遍昨天的HDFS日志再把結(jié)果和實(shí)時(shí)結(jié)果合并。Kappa的做法更簡(jiǎn)單直接把Flink作業(yè)的Kafka消費(fèi)位點(diǎn)重置到上周同一天的0點(diǎn)讓作業(yè)重新跑一遍幾分鐘后就能拿到完整的歷史計(jì)算結(jié)果。數(shù)據(jù)規(guī)模上來之后這體驗(yàn)差別會(huì)越來越明顯。1.2 技術(shù)選型這套方案里的每一個(gè)組件都不是湊數(shù)的ELKFlinkKafka這套組合每一個(gè)組件承擔(dān)的職責(zé)都很清晰沒有一個(gè)是可以砍掉的Kafka是整條鏈路的地基。它承接所有實(shí)時(shí)日志數(shù)據(jù)用分區(qū)機(jī)制提供并行度用offset機(jī)制支撐Flink的exactly-once狀態(tài)恢復(fù)用數(shù)據(jù)保留策略支撐Kappa架構(gòu)的核心——數(shù)據(jù)重放。沒有KafkaFlink的checkpoint恢復(fù)和重放能力就無從談起。Flink是計(jì)算引擎。日志分析的實(shí)時(shí)ETL、指標(biāo)聚合、窗口統(tǒng)計(jì)、異常檢測(cè)都跑在Flink上。選Flink而不是Spark Streaming核心考量是Flink的原生流處理語義、低延遲特性和精確一次exactly-once的狀態(tài)一致性保證。Spark Streaming的micro-batch模式在處理秒級(jí)窗口時(shí)延遲偏高而且批流一體做起來比Flink要費(fèi)勁得多。Elasticsearch負(fù)責(zé)存儲(chǔ)與檢索。清洗后的日志寫入ESKibana負(fù)責(zé)可視化。ES的倒排索引、聚合分析能力天然適合日志場(chǎng)景——業(yè)務(wù)方要查某個(gè)用戶的所有操作記錄、統(tǒng)計(jì)某個(gè)接口的錯(cuò)誤率、分析某個(gè)時(shí)間段的流量走勢(shì)這些都是ES的主場(chǎng)。這套方案能解決的問題邊界也很清楚適合日志量大、實(shí)時(shí)性要求高、需要靈活檢索和分析的場(chǎng)景。如果你的日志量小到單機(jī)就能搞定或者離線分析需求遠(yuǎn)大于實(shí)時(shí)需求那這套方案的復(fù)雜度對(duì)你來說就是純負(fù)擔(dān)。注意Kappa架構(gòu)有個(gè)隱含前提——你的消息中間件必須能保留足夠長時(shí)間的數(shù)據(jù)。如果你遇到的是日志量巨大、Kafka保留窗口只能覆蓋幾個(gè)小時(shí)的場(chǎng)景Kappa的“重放”優(yōu)勢(shì)就沒有了這時(shí)候需要認(rèn)真考慮Lambda或者混合架構(gòu)。這是我踩過一次大坑后得到的體會(huì)。2. 核心組件部署與配置實(shí)戰(zhàn)2.1 Kafka集群參數(shù)規(guī)劃比安裝更重要Kafka集群的規(guī)劃不能只看節(jié)點(diǎn)數(shù)要算清楚吞吐量和存儲(chǔ)的匹配關(guān)系。這里給出一個(gè)實(shí)戰(zhàn)參考假設(shè)單日日志量約200GB日志峰值速率大約是每秒30MB到50MB一般至少需要3個(gè)Kafka節(jié)點(diǎn)每個(gè)節(jié)點(diǎn)掛2塊獨(dú)立數(shù)據(jù)盤做目錄分離。安裝Kafka本身不復(fù)雜網(wǎng)上教程滿天飛。真正的難點(diǎn)在參數(shù)。我挑幾個(gè)踩過坑的配置說# server.properties 核心配置參考 broker.id0 log.dirs/data/kafka-logs-1,/data/kafka-logs-2 num.partitions12 log.retention.hours168 log.segment.bytes1073741824 log.retention.check.interval.ms300000 replica.lag.time.max.ms30000 offsets.topic.replication.factor3 transaction.state.log.replication.factor3 min.insync.replicas2第一條避坑log.segment.bytes默認(rèn)1GB這個(gè)值不用動(dòng)。但要注意log.retention.hours和消息總流量的匹配。我見過有人為了省磁盤把retention設(shè)成24小時(shí)結(jié)果某天Flink作業(yè)掛了一天后恢復(fù)時(shí)發(fā)現(xiàn)Kafka從第20個(gè)小時(shí)開始的數(shù)據(jù)已經(jīng)被清掉了無法完整重放。建議至少保留72小時(shí)留出故障恢復(fù)的窗口。第二條避坑min.insync.replicas2必須設(shè)。如果你的Kafka集群只有3個(gè)節(jié)點(diǎn)副本因子設(shè)為2或者3生產(chǎn)端開啟acksall這樣配置能保證部分節(jié)點(diǎn)故障時(shí)寫入不丟數(shù)據(jù)。這個(gè)參數(shù)不設(shè)生產(chǎn)端配合不當(dāng)會(huì)有丟數(shù)據(jù)的風(fēng)險(xiǎn)。默認(rèn)分區(qū)數(shù)我一般設(shè)成12原因后面講Flink并行度的時(shí)候會(huì)解釋。如果你的Flink作業(yè)并行度很高分區(qū)數(shù)也要跟著提升每個(gè)分區(qū)就是Flink的一個(gè)消費(fèi)并行度來源。分區(qū)數(shù)一旦確定后期擴(kuò)容是要花不少代價(jià)的——從頭新建topic、讓Flink重新消費(fèi)做數(shù)據(jù)遷移。所以初期寧可設(shè)大一點(diǎn)。2.2 Flink部署模式與內(nèi)存配置Flink的部署方式有三種Standalone、YARN Session、YARN Per-Job新版本里推薦Application Mode。實(shí)時(shí)日志分析這種場(chǎng)景我推薦用YARN Session模式。原因很簡(jiǎn)單日志分析任務(wù)不算重型作業(yè)Session模式允許多個(gè)Flink作業(yè)共享一個(gè)集群資源利用率高作業(yè)啟動(dòng)速度快。Per-Job模式每個(gè)作業(yè)啟動(dòng)一個(gè)專用集群隔離性好但資源開銷大。關(guān)于Flink的內(nèi)存配置有一條極其重要的經(jīng)驗(yàn)一定要給Flink設(shè)置獨(dú)立的堆外內(nèi)存和系統(tǒng)內(nèi)存否則默認(rèn)配置在容器環(huán)境下很容易出事。# conf/flink-conf.yaml 關(guān)鍵配置示例 jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 4096m taskmanager.memory.managed.size: 2048m taskmanager.numberOfTaskSlots: 2 parallelism.default: 4 state.backend: rocksdb state.backend.incremental: true checkpointing.interval: 60000state.backend用RocksDB而不是默認(rèn)的HashMap是因?yàn)槿罩痉治鲎鳂I(yè)通常要保存較大規(guī)模的狀態(tài)比如窗口聚合的中間結(jié)果。RocksDB支持增量checkpoint在大狀態(tài)場(chǎng)景下性能要好很多。這里想特別強(qiáng)調(diào)task slot數(shù)量不要盲目設(shè)成和CPU核數(shù)一致。每個(gè)slot上運(yùn)行的任務(wù)要占用內(nèi)存slot太多會(huì)導(dǎo)致堆內(nèi)存溢出slot太少則CPU利用率不足。我在生產(chǎn)環(huán)境使用的經(jīng)驗(yàn)是單機(jī)slot數(shù)量CPU核數(shù)的一半左右比較穩(wěn)妥。2.3 ELK部署別用默認(rèn)配置直接上ELK的部署現(xiàn)在基本都是Docker Compose一把梭。網(wǎng)上搜“elk docker 部署”能搜到一堆模板但默認(rèn)模板直接拿來用會(huì)埋不少雷。先看一個(gè)精簡(jiǎn)的docker-compose版本然后逐個(gè)說坑version: 3.8 services: elasticsearch: image: elasticsearch:7.17.9 environment: - cluster.namees-log-cluster - discovery.typesingle-node - ES_JAVA_OPTS-Xms4g -Xmx4g - bootstrap.memory_locktrue volumes: - es-data:/usr/share/elasticsearch/data ports: - 9200:9200 kibana: image: kibana:7.17.9 environment: - ELASTICSEARCH_HOSTShttp://elasticsearch:9200 - I18N_LOCALEzh-CN ports: - 5601:5601 depends_on: - elasticsearch logstash: image: logstash:7.17.9 volumes: - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf environment: - LS_JAVA_OPTS-Xms2g -Xmx2g depends_on: - elasticsearch第一個(gè)大坑ES的ES_JAVA_OPTS堆內(nèi)存配置。默認(rèn)JVM堆只有2GB日志數(shù)據(jù)一旦上來分片數(shù)又設(shè)得很大幾乎必然OOM。經(jīng)驗(yàn)是按機(jī)器內(nèi)存的50%給ES堆內(nèi)存但不要超過32GB。再往上走JVM的對(duì)象指針壓縮就失效了性能反而下降。第二個(gè)大坑bootstrap.memory_locktrue配合vm.max_map_count的系統(tǒng)參數(shù)。ES需要鎖定內(nèi)存防止交換到磁盤但如果宿主機(jī)沒設(shè)置vm.max_map_countES啟動(dòng)時(shí)會(huì)報(bào)max virtual memory areas vm.max_map_count [65530] is too low。解決辦法是在宿主機(jī)執(zhí)行sudo sysctl -w vm.max_map_count262144第三個(gè)大坑Logstash不裝時(shí)好端端的一加上就瘋狂占內(nèi)存。Logstash默認(rèn)JVM堆1GB處理高吞吐日志時(shí)根本不夠。設(shè)置LS_JAVA_OPTS-Xms2g -Xmx2g是基礎(chǔ)更關(guān)鍵的是別讓Logstash承擔(dān)太重的解析工作——復(fù)雜的grok正則解析會(huì)嚴(yán)重拖慢吞吐能用Flink清洗的字段就丟給FlinkLogstash只做最輕量級(jí)的托運(yùn)。ES索引的生命周期管理ILM是另一個(gè)不能偷懶的點(diǎn)。日志數(shù)據(jù)按天建索引保留30天足夠ILM策略自動(dòng)滾動(dòng)和刪除舊索引省心又防止磁盤被打滿PUT _ilm/policy/log_retention_policy { policy: { phases: { hot: { actions: { rollover: { max_size: 50GB, max_age: 1d } } }, delete: { min_age: 30d, actions: { delete: {} } } } } }3. 實(shí)時(shí)日志分析鏈路的核心實(shí)現(xiàn)3.1 端到端鏈路從日志產(chǎn)生到Kibana圖表這條鏈路我用一個(gè)nginx訪問日志的例子走一遍全流程Filebeat采集每臺(tái)服務(wù)器上部署Filebeat讀取nginx的access.log把每行日志轉(zhuǎn)成JSON消息發(fā)送到Kafka。Kafka緩沖消息按nginx-log這個(gè)topic組織默認(rèn)12個(gè)分區(qū)按服務(wù)器IP或請(qǐng)求路徑做key保證同一來源的日志有序。Flink清洗與計(jì)算消費(fèi)Kafka消息解析出時(shí)間戳、客戶端IP、請(qǐng)求路徑、狀態(tài)碼、響應(yīng)耗時(shí)等字段做ETL清洗然后按1分鐘窗口聚合出各接口的調(diào)用量、P95耗時(shí)、錯(cuò)誤率。ES存儲(chǔ)Flink把清洗后的明細(xì)數(shù)據(jù)寫入nginx-access-log-YYYY.MM.dd索引聚合結(jié)果寫入nginx-access-metric索引。Kibana展示在Kibana里創(chuàng)建Dashboard實(shí)時(shí)展示各接口的吞吐、錯(cuò)誤率趨勢(shì)、TOP訪問IP等。鏈路看起來不長但每一環(huán)的細(xì)節(jié)都能要你命。Filebeat采集端的細(xì)節(jié)要設(shè)置publisher_confirms: true默認(rèn)配置下Filebeat寫Kafka是異步送達(dá)一旦broker端短暫不可用消息就丟了。3.2 Flink作業(yè)讀取、窗口與Processor的完整實(shí)現(xiàn)寫一個(gè)大概的Flink作業(yè)骨架覆蓋日志分析最常見的需求public class LogAnalysisJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); Properties kafkaProps new Properties(); kafkaProps.setProperty(bootstrap.servers, kafka-1:9092,kafka-2:9092,kafka-3:9092); kafkaProps.setProperty(group.id, log-analysis-group); kafkaProps.setProperty(auto.offset.reset, earliest); // 事務(wù)讀配合Flink的checkpoint保證exactly-once FlinkKafkaConsumerString consumer new FlinkKafkaConsumer( nginx-log, new SimpleStringSchema(), kafkaProps ); consumer.setStartFromLatest(); // 首次部署從當(dāng)前時(shí)間開始消費(fèi) DataStreamString rawLogStream env.addSource(consumer); SingleOutputStreamOperatorAccessLog logStream rawLogStream .map(new JsonToAccessLogFunction()) .assignTimestampsAndWatermarks( WatermarkStrategy.AccessLogforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((log, ts) - log.getTimestamp()) ); // 窗口聚合每1分鐘統(tǒng)計(jì)各接口的調(diào)用量、平均耗時(shí)、P95耗時(shí) DataStreamInterfaceMetric metricStream logStream .filter(log - log.getStatus() 200) .keyBy(AccessLog::getApiPath) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new MetricAggregateFunction(), new MetricWindowProcessFunction()); // 寫入ES metricStream.addSink(createElasticsearchSink(nginx-access-metric)); logStream.addSink(createElasticsearchSink(nginx-access-log)); env.execute(nginx-log-analysis); } }這里有幾個(gè)非常關(guān)鍵的實(shí)現(xiàn)細(xì)節(jié)第一EventTime和Watermark必須設(shè)置。日志數(shù)據(jù)的業(yè)務(wù)時(shí)間本身是事件發(fā)生時(shí)間如果直接拿Flink處理時(shí)間來做窗口統(tǒng)計(jì)任何網(wǎng)絡(luò)延遲和反壓都會(huì)導(dǎo)致統(tǒng)計(jì)錨點(diǎn)錯(cuò)亂。設(shè)置forBoundedOutOfOrderness(Duration.ofSeconds(10))允許日志亂序10秒以內(nèi)這個(gè)值要按實(shí)際網(wǎng)絡(luò)環(huán)境調(diào)整設(shè)太大窗口輸出延遲高設(shè)太小丟數(shù)據(jù)。第二聚合函數(shù)里要做狀態(tài)清理。MetricAggregateFunction里保存的就是窗口內(nèi)狀態(tài)的累加器。如果不清理過期key長尾的接口路徑會(huì)持續(xù)占用內(nèi)存。窗口結(jié)束后要主動(dòng)清理狀態(tài)或者用Flink的TTL機(jī)制給狀態(tài)設(shè)置過期時(shí)間。第三ES Sink要設(shè)置冪等寫入。日志場(chǎng)景的冪等最簡(jiǎn)單實(shí)用——ES按_id做upsert。給每條日志生成一個(gè)MD5(時(shí)間戳 日志原文)作為文檔ID這樣即使Flink作業(yè)發(fā)生故障重放同一批次數(shù)據(jù)重復(fù)寫入時(shí)也會(huì)因?yàn)镮D相同被覆蓋不會(huì)產(chǎn)生重復(fù)文檔。3.3 寫入ES的調(diào)優(yōu)細(xì)節(jié)bulk是王道ES Sink的性能是整個(gè)鏈路的瓶頸之一。默認(rèn)的ES connector寫入是逐條、同步的日志量大時(shí)吞吐根本扛不住。我建議所有生產(chǎn)環(huán)境的ES Sink都開啟bulk模式private static ElasticsearchSinkAccessLog createElasticsearchSink(String indexName) { ListHttpHost httpHosts new ArrayList(); httpHosts.add(new HttpHost(es-1, 9200, http)); ElasticsearchSinkFunctionAccessLog sinkFunction new ElasticsearchSinkFunctionAccessLog() { Override public void process(AccessLog log, RuntimeContext ctx, RequestIndexer indexer) { MapString, Object json new HashMap(); json.put(apiPath, log.getApiPath()); json.put(status, log.getStatus()); json.put(costMs, log.getCostMs()); IndexRequest request Requests.indexRequest() .index(indexName) .id(log.generateId()) .source(json); indexer.add(request); } }; return new ElasticsearchSink.Builder(httpHosts, sinkFunction) .setBulkFlushMaxActions(5000) .setBulkFlushMaxSizeMb(100) .setBulkFlushInterval(5000) .build(); }我把bulkFlushMaxActions設(shè)為5000maxSizeMb設(shè)為100MBflushInterval設(shè)為5秒。這幾個(gè)值是根據(jù)ES的寫入吞吐實(shí)測(cè)出來的平衡點(diǎn)。bulk太頻繁會(huì)增加ES的索引壓力bulk太少則不能充分合并寫入請(qǐng)求。ES服務(wù)端的兩個(gè)參數(shù)對(duì)日志場(chǎng)景極其關(guān)鍵PUT /_cluster/settings { transient: { indices.memory.index_buffer_size: 20%, indices.requests.cache.size: 5% } }index_buffer_size決定ES在落到磁盤前能在內(nèi)存里攢多少數(shù)據(jù)20%是官方建議值別貪大太大容易OOM。requests.cache.size只在大量重復(fù)聚合場(chǎng)景下有收益日志檢索場(chǎng)景設(shè)置5%足夠了。4. 常見問題排查與調(diào)優(yōu)實(shí)錄4.1 Kafka消息延遲高問題可能不在Kafka“Kafka延遲高”是我被問過最多的問題。排查這類問題有一個(gè)黃金法則先看生產(chǎn)端再看消費(fèi)端最后看Broker。很多時(shí)候Kafka自己根本沒毛病。最典型的場(chǎng)景是業(yè)務(wù)方反饋日志從產(chǎn)生到出現(xiàn)在Kibana里延遲了十幾分鐘。查Kafka broker的CPU和網(wǎng)絡(luò)都正常topic的分區(qū)數(shù)12個(gè)消費(fèi)端Flink作業(yè)各并行度也正常那問題大概率出在三個(gè)方面生產(chǎn)端batch.size和linger.ms搭配不佳。Kafka生產(chǎn)端默認(rèn)batch.size16KBlinger.ms0。批量太小、等待時(shí)間太短會(huì)導(dǎo)致每條消息都單獨(dú)發(fā)一次網(wǎng)絡(luò)請(qǐng)求網(wǎng)絡(luò)往返消耗遠(yuǎn)大于發(fā)送數(shù)據(jù)本身。把batch.size適當(dāng)調(diào)大比如64KBlinger.ms設(shè)為5到10毫秒Kafka吞吐會(huì)有立竿見影的提升。注意linger.ms不是延遲發(fā)送多少毫秒的意思而是等待攢夠一個(gè)批次的最長等待時(shí)間5毫秒級(jí)別的設(shè)置對(duì)實(shí)時(shí)性幾乎無感。消費(fèi)端fetch.max.bytes設(shè)置過小。這是另一個(gè)容易被忽略的點(diǎn)。Flink的Kafka消費(fèi)者默認(rèn)fetch.max.bytes50MB但單個(gè)分區(qū)的fetch.max.bytes默認(rèn)是1MB。如果你設(shè)置了12個(gè)分區(qū)每個(gè)分區(qū)的消費(fèi)并發(fā)一次fetch能拉取的數(shù)據(jù)量可能撐不滿網(wǎng)絡(luò)帶寬。日志場(chǎng)景的消費(fèi)速度頻繁被這個(gè)參數(shù)拖后腿。建議顯式設(shè)置為Properties kafkaProps new Properties(); kafkaProps.setProperty(FlinkKafkaConsumer.KEY_FETCH_MAX_BYTES, 52428800);Flink作業(yè)存在反壓。這是最多發(fā)的情況。日志高峰期數(shù)據(jù)量暴增Flink的源端消費(fèi)不過來下游ES寫入跟不上整個(gè)鏈路卡住。最直接的表現(xiàn)是Kafka的consumer lag持續(xù)增長。排查方法是看Flink UI上每個(gè)算子是否有背壓告警或者直接看Kafka consumer group的lag指標(biāo)。如果確認(rèn)是ES寫入瓶頸除了前面提到的bulk調(diào)優(yōu)外還可以給ES增加數(shù)據(jù)節(jié)點(diǎn)或者檢查ES索引的分片數(shù)量是否過多——分片過多會(huì)導(dǎo)致每寫一條數(shù)據(jù)都要和所有分片協(xié)調(diào)性能反而不升反降。4.2 Flink的JDBC連接器異常幾乎都是連接池配置問題Flink寫MySQL或別的數(shù)據(jù)庫報(bào)連接器異常我排查過的case里八九成是連接池相關(guān)配置不當(dāng)。典型報(bào)錯(cuò)是Could not initialize class org.apache.flink.connector.jdbc.table.JdbcDialect或者Caused by: java.sql.SQLException: Cannot create PoolableConnectionFactory排查思路按順序走第一檢查驅(qū)動(dòng)版本和Flink版本是否匹配。Flink 1.15以上用JDBC Connector 2.x底層數(shù)據(jù)庫驅(qū)動(dòng)如果太舊會(huì)出現(xiàn)不兼容的異常。這類問題去搜“flink jdbc connector異?！蹦苷业讲簧侔咐夥ù蠖嗍巧?jí)驅(qū)動(dòng)版本。第二檢查數(shù)據(jù)庫連接數(shù)限制。日志分析場(chǎng)景給Flink配置連接池大小不是越大越好而是取決于下游數(shù)據(jù)庫的max_connections。比如MySQL默認(rèn)max_connections151你給Flink配50個(gè)連接還要考慮別的服務(wù)很可能直接把數(shù)據(jù)庫打爆。穩(wěn)妥做法Flink的JDBC連接池大小不要超過數(shù)據(jù)庫最大連接數(shù)的20%。第三檢查checkpoint恢復(fù)后的連接狀態(tài)。Flink任務(wù)重啟恢復(fù)時(shí)舊連接可能已經(jīng)失效需要設(shè)置JdbcExecutionOptions的自動(dòng)重連參數(shù)。我在Flink里一般這樣配JdbcExecutionOptions.builder() .withBatchSize(5000) .withBatchIntervalMs(2000) .withMaxRetries(3) .build();4.3 ES寫入報(bào)錯(cuò)與Kafka的InvalidReceiveException日志鏈路的另一個(gè)高頻故障是啟動(dòng)時(shí)ES集群還沒就緒Flink的Sink已經(jīng)開始寫入報(bào)各種節(jié)點(diǎn)不可用、shard lock異常。規(guī)避方法是在鏈路啟動(dòng)前做一次健康檢查——確認(rèn)ES的/_cluster/health返回的statusgreen或者至少yellow再啟動(dòng)Flink作業(yè)。生產(chǎn)環(huán)境建議把健康檢查腳本寫成shell腳本在CI/CD流水線里檢查。還有一個(gè)必須認(rèn)識(shí)清楚的經(jīng)典Kafka報(bào)錯(cuò)org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size 647204481 larger than 100000000)我第一看到這個(gè)報(bào)錯(cuò)時(shí)也懵了后來排查清楚才知道兩個(gè)原因最常見一是客戶端配置的receive.buffer.bytes和broker端不匹配。某次我排查的時(shí)候發(fā)現(xiàn)Flink客戶端的receive.buffer.bytes被設(shè)成了100MB而broker端的socket.request.max.bytes默認(rèn)只有100MB。當(dāng)單條消息大小接近這個(gè)值時(shí)會(huì)觸法這個(gè)異常。解決方案是把message.max.bytes和socket.request.max.bytes在broker端和客戶端都調(diào)大并保持匹配。二是客戶端反序列化框架的版本不一致。Kafka客戶端把字節(jié)流反序列化時(shí)會(huì)校驗(yàn)FrameSize版本不一致或被污染的數(shù)據(jù)流就會(huì)出現(xiàn)這個(gè)異常。排查這種報(bào)錯(cuò)的標(biāo)準(zhǔn)姿勢(shì)是先看客戶端配置再看broker端配置逐個(gè)對(duì)齊同時(shí)檢查兩端Kafka版本是否一致。4.4 Kafka消費(fèi)端多線程如何保證消息順序性日志分析場(chǎng)景對(duì)全局順序的要求通常不高但如果你要處理某個(gè)用戶的完整操作鏈路同一用戶的操作日志必須保證順序。Kafka保證順序性的前提是同一分區(qū)內(nèi)的消息按offset遞增順序消費(fèi)而相同key的消息會(huì)被路由到同一個(gè)分區(qū)。Flink消費(fèi)Kafka時(shí)默認(rèn)以partition為單位做并行消費(fèi)Flink內(nèi)部每個(gè)partition對(duì)應(yīng)一個(gè)subtask天然保證同一個(gè)分區(qū)內(nèi)的消息按順序處理。問題出在多線程處理下游的環(huán)節(jié)如果Flink算子內(nèi)部用了線程池并發(fā)處理消息順序就亂了。我踩過這個(gè)坑后總結(jié)了三個(gè)保序方案按推薦程度排序方案一提高Flink并行度但保證相同key的數(shù)據(jù)進(jìn)同一個(gè)分區(qū)。Flink上游Kafka Source并行度等于Kafka分區(qū)數(shù)保證相同key的消息進(jìn)入同一個(gè)子任務(wù)即可保序。這個(gè)方案最干凈前提是你的并行度要求和分區(qū)數(shù)匹配。方案二關(guān)鍵算子內(nèi)部用單線程。如果你不得不在算子內(nèi)部并發(fā)處理比如外部IO較慢那就要隱藏分區(qū)key讓該key的所有數(shù)據(jù)都被路由到同一個(gè)線程。做法是自定義一個(gè)KeyedProcessFunction內(nèi)部用單線程處理每個(gè)key的數(shù)據(jù)。方案三放棄全局嚴(yán)格順序用事件時(shí)間水位線兜底。日志場(chǎng)景里95%的“順序性問題”其實(shí)可以用窗口和事件時(shí)間優(yōu)雅解決。Flink的Watermark機(jī)制允許一定程度的亂序只要延遲在容忍范圍內(nèi)計(jì)算結(jié)果就是正確的。最后提醒一個(gè)特別容易犯的錯(cuò)誤如果你為了保序而把所有數(shù)據(jù)都發(fā)送到同一個(gè)分區(qū)那你等于放棄了Kafka的并行能力整個(gè)鏈路的吞吐會(huì)驟降。生產(chǎn)環(huán)境優(yōu)先選方案一把保序收斂到key級(jí)別而不是全局。5. 這套方案的后續(xù)擴(kuò)展方向剛才提到的這些都還只是實(shí)時(shí)日志分析的基線能力。鏈路搭好之后往上擴(kuò)展的空間非常大——比如把Flink的Cep模式匹配能力接進(jìn)來做異常行為實(shí)時(shí)告警又比如把日志指標(biāo)輸出到Prometheus用Grafana做基礎(chǔ)監(jiān)控還比如在Flink里接入OpenMetadata自動(dòng)采集Flink作業(yè)的血緣關(guān)系讓數(shù)據(jù)資產(chǎn)的元數(shù)據(jù)跟上實(shí)時(shí)的節(jié)奏。我個(gè)人的體會(huì)是日志分析平臺(tái)永遠(yuǎn)不是靜態(tài)工程它更像一個(gè)不斷生長的基座不停接入新的數(shù)據(jù)源、新的分析維度、新的下游系統(tǒng)。架構(gòu)選型時(shí)如果沒留出擴(kuò)展的余地后面每一次新需求都要傷筋動(dòng)骨。而這套ELKFlinkKafka的組合擴(kuò)展性恰恰是我在工程實(shí)踐中體會(huì)最深的一點(diǎn)。 ELK+Flink+Kafka：實(shí)時(shí)日志分析平臺(tái)的Kappa架構(gòu)實(shí)戰(zhàn)
1. 項(xiàng)目整體設(shè)計(jì)與Kappa架構(gòu)選型背后的邏輯1.1 為什么是Kappa而不是Lambda先說結(jié)論如果你現(xiàn)在還要為一個(gè)新項(xiàng)目搭建實(shí)時(shí)日志分析平臺(tái)Lambda架構(gòu)大概率已經(jīng)不是最優(yōu)解了。Kappa架構(gòu)的核心思想非常樸素——把所有數(shù)據(jù)都當(dāng)作流來處理用一個(gè)引擎同時(shí)支撐實(shí)時(shí)計(jì)算和歷史數(shù)據(jù)重放不需要像Lambda那樣為批處理和流處理各維護(hù)一套代碼。Lambda架構(gòu)給人挖的坑我太有體會(huì)了。流批兩套代碼意味著兩套邏輯、兩套部署、兩套運(yùn)維最痛苦的是當(dāng)你要修一個(gè)bug或者加一個(gè)字段時(shí)要在兩個(gè)項(xiàng)目里分別改一遍然后還得對(duì)兩邊的計(jì)算結(jié)果做合并和校驗(yàn)。日志分析這個(gè)場(chǎng)景尤其尷尬日志數(shù)據(jù)本質(zhì)上就是一條條不斷產(chǎn)生的事件流你非要用批處理框架對(duì)一份靜態(tài)文件反復(fù)跑批屬于脫褲子放屁。Kappa的底氣來自Kafka的持久化和重放能力。Kafka可以保留全量日志數(shù)據(jù)通常按天或按容量設(shè)置retention當(dāng)業(yè)務(wù)方需要重新計(jì)算某個(gè)時(shí)間窗口的指標(biāo)時(shí)我們只要把Kafka的消費(fèi)位點(diǎn)重置到那個(gè)時(shí)間點(diǎn)之前再用Flink從那個(gè)位點(diǎn)重新消費(fèi)、重新計(jì)算結(jié)果寫到新的Elasticsearch索引里就行。整個(gè)過程不需要啟動(dòng)任何批處理任務(wù)也不需要寫一套MapReduce代碼。舉一個(gè)具體例子某天凌晨線上有一個(gè)支付接口的調(diào)用量異常飆升業(yè)務(wù)方想對(duì)比今天早高峰和上周同一天的數(shù)據(jù)。Lambda架構(gòu)的做法是臨時(shí)寫一個(gè)Hive SQL跑一遍昨天的HDFS日志再把結(jié)果和實(shí)時(shí)結(jié)果合并。Kappa的做法更簡(jiǎn)單直接把Flink作業(yè)的Kafka消費(fèi)位點(diǎn)重置到上周同一天的0點(diǎn)讓作業(yè)重新跑一遍幾分鐘后就能拿到完整的歷史計(jì)算結(jié)果。數(shù)據(jù)規(guī)模上來之后這體驗(yàn)差別會(huì)越來越明顯。1.2 技術(shù)選型這套方案里的每一個(gè)組件都不是湊數(shù)的ELKFlinkKafka這套組合每一個(gè)組件承擔(dān)的職責(zé)都很清晰沒有一個(gè)是可以砍掉的Kafka是整條鏈路的地基。它承接所有實(shí)時(shí)日志數(shù)據(jù)用分區(qū)機(jī)制提供并行度用offset機(jī)制支撐Flink的exactly-once狀態(tài)恢復(fù)用數(shù)據(jù)保留策略支撐Kappa架構(gòu)的核心——數(shù)據(jù)重放。沒有KafkaFlink的checkpoint恢復(fù)和重放能力就無從談起。Flink是計(jì)算引擎。日志分析的實(shí)時(shí)ETL、指標(biāo)聚合、窗口統(tǒng)計(jì)、異常檢測(cè)都跑在Flink上。選Flink而不是Spark Streaming核心考量是Flink的原生流處理語義、低延遲特性和精確一次exactly-once的狀態(tài)一致性保證。Spark Streaming的micro-batch模式在處理秒級(jí)窗口時(shí)延遲偏高而且批流一體做起來比Flink要費(fèi)勁得多。Elasticsearch負(fù)責(zé)存儲(chǔ)與檢索。清洗后的日志寫入ESKibana負(fù)責(zé)可視化。ES的倒排索引、聚合分析能力天然適合日志場(chǎng)景——業(yè)務(wù)方要查某個(gè)用戶的所有操作記錄、統(tǒng)計(jì)某個(gè)接口的錯(cuò)誤率、分析某個(gè)時(shí)間段的流量走勢(shì)這些都是ES的主場(chǎng)。這套方案能解決的問題邊界也很清楚適合日志量大、實(shí)時(shí)性要求高、需要靈活檢索和分析的場(chǎng)景。如果你的日志量小到單機(jī)就能搞定或者離線分析需求遠(yuǎn)大于實(shí)時(shí)需求那這套方案的復(fù)雜度對(duì)你來說就是純負(fù)擔(dān)。注意Kappa架構(gòu)有個(gè)隱含前提——你的消息中間件必須能保留足夠長時(shí)間的數(shù)據(jù)。如果你遇到的是日志量巨大、Kafka保留窗口只能覆蓋幾個(gè)小時(shí)的場(chǎng)景Kappa的“重放”優(yōu)勢(shì)就沒有了這時(shí)候需要認(rèn)真考慮Lambda或者混合架構(gòu)。這是我踩過一次大坑后得到的體會(huì)。2. 核心組件部署與配置實(shí)戰(zhàn)2.1 Kafka集群參數(shù)規(guī)劃比安裝更重要Kafka集群的規(guī)劃不能只看節(jié)點(diǎn)數(shù)要算清楚吞吐量和存儲(chǔ)的匹配關(guān)系。這里給出一個(gè)實(shí)戰(zhàn)參考假設(shè)單日日志量約200GB日志峰值速率大約是每秒30MB到50MB一般至少需要3個(gè)Kafka節(jié)點(diǎn)每個(gè)節(jié)點(diǎn)掛2塊獨(dú)立數(shù)據(jù)盤做目錄分離。安裝Kafka本身不復(fù)雜網(wǎng)上教程滿天飛。真正的難點(diǎn)在參數(shù)。我挑幾個(gè)踩過坑的配置說# server.properties 核心配置參考 broker.id0 log.dirs/data/kafka-logs-1,/data/kafka-logs-2 num.partitions12 log.retention.hours168 log.segment.bytes1073741824 log.retention.check.interval.ms300000 replica.lag.time.max.ms30000 offsets.topic.replication.factor3 transaction.state.log.replication.factor3 min.insync.replicas2第一條避坑log.segment.bytes默認(rèn)1GB這個(gè)值不用動(dòng)。但要注意log.retention.hours和消息總流量的匹配。我見過有人為了省磁盤把retention設(shè)成24小時(shí)結(jié)果某天Flink作業(yè)掛了一天后恢復(fù)時(shí)發(fā)現(xiàn)Kafka從第20個(gè)小時(shí)開始的數(shù)據(jù)已經(jīng)被清掉了無法完整重放。建議至少保留72小時(shí)留出故障恢復(fù)的窗口。第二條避坑min.insync.replicas2必須設(shè)。如果你的Kafka集群只有3個(gè)節(jié)點(diǎn)副本因子設(shè)為2或者3生產(chǎn)端開啟acksall這樣配置能保證部分節(jié)點(diǎn)故障時(shí)寫入不丟數(shù)據(jù)。這個(gè)參數(shù)不設(shè)生產(chǎn)端配合不當(dāng)會(huì)有丟數(shù)據(jù)的風(fēng)險(xiǎn)。默認(rèn)分區(qū)數(shù)我一般設(shè)成12原因后面講Flink并行度的時(shí)候會(huì)解釋。如果你的Flink作業(yè)并行度很高分區(qū)數(shù)也要跟著提升每個(gè)分區(qū)就是Flink的一個(gè)消費(fèi)并行度來源。分區(qū)數(shù)一旦確定后期擴(kuò)容是要花不少代價(jià)的——從頭新建topic、讓Flink重新消費(fèi)做數(shù)據(jù)遷移。所以初期寧可設(shè)大一點(diǎn)。2.2 Flink部署模式與內(nèi)存配置Flink的部署方式有三種Standalone、YARN Session、YARN Per-Job新版本里推薦Application Mode。實(shí)時(shí)日志分析這種場(chǎng)景我推薦用YARN Session模式。原因很簡(jiǎn)單日志分析任務(wù)不算重型作業(yè)Session模式允許多個(gè)Flink作業(yè)共享一個(gè)集群資源利用率高作業(yè)啟動(dòng)速度快。Per-Job模式每個(gè)作業(yè)啟動(dòng)一個(gè)專用集群隔離性好但資源開銷大。關(guān)于Flink的內(nèi)存配置有一條極其重要的經(jīng)驗(yàn)一定要給Flink設(shè)置獨(dú)立的堆外內(nèi)存和系統(tǒng)內(nèi)存否則默認(rèn)配置在容器環(huán)境下很容易出事。# conf/flink-conf.yaml 關(guān)鍵配置示例 jobmanager.memory.process.size: 2048m taskmanager.memory.process.size: 4096m taskmanager.memory.managed.size: 2048m taskmanager.numberOfTaskSlots: 2 parallelism.default: 4 state.backend: rocksdb state.backend.incremental: true checkpointing.interval: 60000state.backend用RocksDB而不是默認(rèn)的HashMap是因?yàn)槿罩痉治鲎鳂I(yè)通常要保存較大規(guī)模的狀態(tài)比如窗口聚合的中間結(jié)果。RocksDB支持增量checkpoint在大狀態(tài)場(chǎng)景下性能要好很多。這里想特別強(qiáng)調(diào)task slot數(shù)量不要盲目設(shè)成和CPU核數(shù)一致。每個(gè)slot上運(yùn)行的任務(wù)要占用內(nèi)存slot太多會(huì)導(dǎo)致堆內(nèi)存溢出slot太少則CPU利用率不足。我在生產(chǎn)環(huán)境使用的經(jīng)驗(yàn)是單機(jī)slot數(shù)量CPU核數(shù)的一半左右比較穩(wěn)妥。2.3 ELK部署別用默認(rèn)配置直接上ELK的部署現(xiàn)在基本都是Docker Compose一把梭。網(wǎng)上搜“elk docker 部署”能搜到一堆模板但默認(rèn)模板直接拿來用會(huì)埋不少雷。先看一個(gè)精簡(jiǎn)的docker-compose版本然后逐個(gè)說坑version: 3.8 services: elasticsearch: image: elasticsearch:7.17.9 environment: - cluster.namees-log-cluster - discovery.typesingle-node - ES_JAVA_OPTS-Xms4g -Xmx4g - bootstrap.memory_locktrue volumes: - es-data:/usr/share/elasticsearch/data ports: - 9200:9200 kibana: image: kibana:7.17.9 environment: - ELASTICSEARCH_HOSTShttp://elasticsearch:9200 - I18N_LOCALEzh-CN ports: - 5601:5601 depends_on: - elasticsearch logstash: image: logstash:7.17.9 volumes: - ./logstash.conf:/usr/share/logstash/pipeline/logstash.conf environment: - LS_JAVA_OPTS-Xms2g -Xmx2g depends_on: - elasticsearch第一個(gè)大坑ES的ES_JAVA_OPTS堆內(nèi)存配置。默認(rèn)JVM堆只有2GB日志數(shù)據(jù)一旦上來分片數(shù)又設(shè)得很大幾乎必然OOM。經(jīng)驗(yàn)是按機(jī)器內(nèi)存的50%給ES堆內(nèi)存但不要超過32GB。再往上走JVM的對(duì)象指針壓縮就失效了性能反而下降。第二個(gè)大坑bootstrap.memory_locktrue配合vm.max_map_count的系統(tǒng)參數(shù)。ES需要鎖定內(nèi)存防止交換到磁盤但如果宿主機(jī)沒設(shè)置vm.max_map_countES啟動(dòng)時(shí)會(huì)報(bào)max virtual memory areas vm.max_map_count [65530] is too low。解決辦法是在宿主機(jī)執(zhí)行sudo sysctl -w vm.max_map_count262144第三個(gè)大坑Logstash不裝時(shí)好端端的一加上就瘋狂占內(nèi)存。Logstash默認(rèn)JVM堆1GB處理高吞吐日志時(shí)根本不夠。設(shè)置LS_JAVA_OPTS-Xms2g -Xmx2g是基礎(chǔ)更關(guān)鍵的是別讓Logstash承擔(dān)太重的解析工作——復(fù)雜的grok正則解析會(huì)嚴(yán)重拖慢吞吐能用Flink清洗的字段就丟給FlinkLogstash只做最輕量級(jí)的托運(yùn)。ES索引的生命周期管理ILM是另一個(gè)不能偷懶的點(diǎn)。日志數(shù)據(jù)按天建索引保留30天足夠ILM策略自動(dòng)滾動(dòng)和刪除舊索引省心又防止磁盤被打滿PUT _ilm/policy/log_retention_policy { policy: { phases: { hot: { actions: { rollover: { max_size: 50GB, max_age: 1d } } }, delete: { min_age: 30d, actions: { delete: {} } } } } }3. 實(shí)時(shí)日志分析鏈路的核心實(shí)現(xiàn)3.1 端到端鏈路從日志產(chǎn)生到Kibana圖表這條鏈路我用一個(gè)nginx訪問日志的例子走一遍全流程Filebeat采集每臺(tái)服務(wù)器上部署Filebeat讀取nginx的access.log把每行日志轉(zhuǎn)成JSON消息發(fā)送到Kafka。Kafka緩沖消息按nginx-log這個(gè)topic組織默認(rèn)12個(gè)分區(qū)按服務(wù)器IP或請(qǐng)求路徑做key保證同一來源的日志有序。Flink清洗與計(jì)算消費(fèi)Kafka消息解析出時(shí)間戳、客戶端IP、請(qǐng)求路徑、狀態(tài)碼、響應(yīng)耗時(shí)等字段做ETL清洗然后按1分鐘窗口聚合出各接口的調(diào)用量、P95耗時(shí)、錯(cuò)誤率。ES存儲(chǔ)Flink把清洗后的明細(xì)數(shù)據(jù)寫入nginx-access-log-YYYY.MM.dd索引聚合結(jié)果寫入nginx-access-metric索引。Kibana展示在Kibana里創(chuàng)建Dashboard實(shí)時(shí)展示各接口的吞吐、錯(cuò)誤率趨勢(shì)、TOP訪問IP等。鏈路看起來不長但每一環(huán)的細(xì)節(jié)都能要你命。Filebeat采集端的細(xì)節(jié)要設(shè)置publisher_confirms: true默認(rèn)配置下Filebeat寫Kafka是異步送達(dá)一旦broker端短暫不可用消息就丟了。3.2 Flink作業(yè)讀取、窗口與Processor的完整實(shí)現(xiàn)寫一個(gè)大概的Flink作業(yè)骨架覆蓋日志分析最常見的需求public class LogAnalysisJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); Properties kafkaProps new Properties(); kafkaProps.setProperty(bootstrap.servers, kafka-1:9092,kafka-2:9092,kafka-3:9092); kafkaProps.setProperty(group.id, log-analysis-group); kafkaProps.setProperty(auto.offset.reset, earliest); // 事務(wù)讀配合Flink的checkpoint保證exactly-once FlinkKafkaConsumerString consumer new FlinkKafkaConsumer( nginx-log, new SimpleStringSchema(), kafkaProps ); consumer.setStartFromLatest(); // 首次部署從當(dāng)前時(shí)間開始消費(fèi) DataStreamString rawLogStream env.addSource(consumer); SingleOutputStreamOperatorAccessLog logStream rawLogStream .map(new JsonToAccessLogFunction()) .assignTimestampsAndWatermarks( WatermarkStrategy.AccessLogforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((log, ts) - log.getTimestamp()) ); // 窗口聚合每1分鐘統(tǒng)計(jì)各接口的調(diào)用量、平均耗時(shí)、P95耗時(shí) DataStreamInterfaceMetric metricStream logStream .filter(log - log.getStatus() 200) .keyBy(AccessLog::getApiPath) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new MetricAggregateFunction(), new MetricWindowProcessFunction()); // 寫入ES metricStream.addSink(createElasticsearchSink(nginx-access-metric)); logStream.addSink(createElasticsearchSink(nginx-access-log)); env.execute(nginx-log-analysis); } }這里有幾個(gè)非常關(guān)鍵的實(shí)現(xiàn)細(xì)節(jié)第一EventTime和Watermark必須設(shè)置。日志數(shù)據(jù)的業(yè)務(wù)時(shí)間本身是事件發(fā)生時(shí)間如果直接拿Flink處理時(shí)間來做窗口統(tǒng)計(jì)任何網(wǎng)絡(luò)延遲和反壓都會(huì)導(dǎo)致統(tǒng)計(jì)錨點(diǎn)錯(cuò)亂。設(shè)置forBoundedOutOfOrderness(Duration.ofSeconds(10))允許日志亂序10秒以內(nèi)這個(gè)值要按實(shí)際網(wǎng)絡(luò)環(huán)境調(diào)整設(shè)太大窗口輸出延遲高設(shè)太小丟數(shù)據(jù)。第二聚合函數(shù)里要做狀態(tài)清理。MetricAggregateFunction里保存的就是窗口內(nèi)狀態(tài)的累加器。如果不清理過期key長尾的接口路徑會(huì)持續(xù)占用內(nèi)存。窗口結(jié)束后要主動(dòng)清理狀態(tài)或者用Flink的TTL機(jī)制給狀態(tài)設(shè)置過期時(shí)間。第三ES Sink要設(shè)置冪等寫入。日志場(chǎng)景的冪等最簡(jiǎn)單實(shí)用——ES按_id做upsert。給每條日志生成一個(gè)MD5(時(shí)間戳 日志原文)作為文檔ID這樣即使Flink作業(yè)發(fā)生故障重放同一批次數(shù)據(jù)重復(fù)寫入時(shí)也會(huì)因?yàn)镮D相同被覆蓋不會(huì)產(chǎn)生重復(fù)文檔。3.3 寫入ES的調(diào)優(yōu)細(xì)節(jié)bulk是王道ES Sink的性能是整個(gè)鏈路的瓶頸之一。默認(rèn)的ES connector寫入是逐條、同步的日志量大時(shí)吞吐根本扛不住。我建議所有生產(chǎn)環(huán)境的ES Sink都開啟bulk模式private static ElasticsearchSinkAccessLog createElasticsearchSink(String indexName) { ListHttpHost httpHosts new ArrayList(); httpHosts.add(new HttpHost(es-1, 9200, http)); ElasticsearchSinkFunctionAccessLog sinkFunction new ElasticsearchSinkFunctionAccessLog() { Override public void process(AccessLog log, RuntimeContext ctx, RequestIndexer indexer) { MapString, Object json new HashMap(); json.put(apiPath, log.getApiPath()); json.put(status, log.getStatus()); json.put(costMs, log.getCostMs()); IndexRequest request Requests.indexRequest() .index(indexName) .id(log.generateId()) .source(json); indexer.add(request); } }; return new ElasticsearchSink.Builder(httpHosts, sinkFunction) .setBulkFlushMaxActions(5000) .setBulkFlushMaxSizeMb(100) .setBulkFlushInterval(5000) .build(); }我把bulkFlushMaxActions設(shè)為5000maxSizeMb設(shè)為100MBflushInterval設(shè)為5秒。這幾個(gè)值是根據(jù)ES的寫入吞吐實(shí)測(cè)出來的平衡點(diǎn)。bulk太頻繁會(huì)增加ES的索引壓力bulk太少則不能充分合并寫入請(qǐng)求。ES服務(wù)端的兩個(gè)參數(shù)對(duì)日志場(chǎng)景極其關(guān)鍵PUT /_cluster/settings { transient: { indices.memory.index_buffer_size: 20%, indices.requests.cache.size: 5% } }index_buffer_size決定ES在落到磁盤前能在內(nèi)存里攢多少數(shù)據(jù)20%是官方建議值別貪大太大容易OOM。requests.cache.size只在大量重復(fù)聚合場(chǎng)景下有收益日志檢索場(chǎng)景設(shè)置5%足夠了。4. 常見問題排查與調(diào)優(yōu)實(shí)錄4.1 Kafka消息延遲高問題可能不在Kafka“Kafka延遲高”是我被問過最多的問題。排查這類問題有一個(gè)黃金法則先看生產(chǎn)端再看消費(fèi)端最后看Broker。很多時(shí)候Kafka自己根本沒毛病。最典型的場(chǎng)景是業(yè)務(wù)方反饋日志從產(chǎn)生到出現(xiàn)在Kibana里延遲了十幾分鐘。查Kafka broker的CPU和網(wǎng)絡(luò)都正常topic的分區(qū)數(shù)12個(gè)消費(fèi)端Flink作業(yè)各并行度也正常那問題大概率出在三個(gè)方面生產(chǎn)端batch.size和linger.ms搭配不佳。Kafka生產(chǎn)端默認(rèn)batch.size16KBlinger.ms0。批量太小、等待時(shí)間太短會(huì)導(dǎo)致每條消息都單獨(dú)發(fā)一次網(wǎng)絡(luò)請(qǐng)求網(wǎng)絡(luò)往返消耗遠(yuǎn)大于發(fā)送數(shù)據(jù)本身。把batch.size適當(dāng)調(diào)大比如64KBlinger.ms設(shè)為5到10毫秒Kafka吞吐會(huì)有立竿見影的提升。注意linger.ms不是延遲發(fā)送多少毫秒的意思而是等待攢夠一個(gè)批次的最長等待時(shí)間5毫秒級(jí)別的設(shè)置對(duì)實(shí)時(shí)性幾乎無感。消費(fèi)端fetch.max.bytes設(shè)置過小。這是另一個(gè)容易被忽略的點(diǎn)。Flink的Kafka消費(fèi)者默認(rèn)fetch.max.bytes50MB但單個(gè)分區(qū)的fetch.max.bytes默認(rèn)是1MB。如果你設(shè)置了12個(gè)分區(qū)每個(gè)分區(qū)的消費(fèi)并發(fā)一次fetch能拉取的數(shù)據(jù)量可能撐不滿網(wǎng)絡(luò)帶寬。日志場(chǎng)景的消費(fèi)速度頻繁被這個(gè)參數(shù)拖后腿。建議顯式設(shè)置為Properties kafkaProps new Properties(); kafkaProps.setProperty(FlinkKafkaConsumer.KEY_FETCH_MAX_BYTES, 52428800);Flink作業(yè)存在反壓。這是最多發(fā)的情況。日志高峰期數(shù)據(jù)量暴增Flink的源端消費(fèi)不過來下游ES寫入跟不上整個(gè)鏈路卡住。最直接的表現(xiàn)是Kafka的consumer lag持續(xù)增長。排查方法是看Flink UI上每個(gè)算子是否有背壓告警或者直接看Kafka consumer group的lag指標(biāo)。如果確認(rèn)是ES寫入瓶頸除了前面提到的bulk調(diào)優(yōu)外還可以給ES增加數(shù)據(jù)節(jié)點(diǎn)或者檢查ES索引的分片數(shù)量是否過多——分片過多會(huì)導(dǎo)致每寫一條數(shù)據(jù)都要和所有分片協(xié)調(diào)性能反而不升反降。4.2 Flink的JDBC連接器異常幾乎都是連接池配置問題Flink寫MySQL或別的數(shù)據(jù)庫報(bào)連接器異常我排查過的case里八九成是連接池相關(guān)配置不當(dāng)。典型報(bào)錯(cuò)是Could not initialize class org.apache.flink.connector.jdbc.table.JdbcDialect或者Caused by: java.sql.SQLException: Cannot create PoolableConnectionFactory排查思路按順序走第一檢查驅(qū)動(dòng)版本和Flink版本是否匹配。Flink 1.15以上用JDBC Connector 2.x底層數(shù)據(jù)庫驅(qū)動(dòng)如果太舊會(huì)出現(xiàn)不兼容的異常。這類問題去搜“flink jdbc connector異?！蹦苷业讲簧侔咐夥ù蠖嗍巧?jí)驅(qū)動(dòng)版本。第二檢查數(shù)據(jù)庫連接數(shù)限制。日志分析場(chǎng)景給Flink配置連接池大小不是越大越好而是取決于下游數(shù)據(jù)庫的max_connections。比如MySQL默認(rèn)max_connections151你給Flink配50個(gè)連接還要考慮別的服務(wù)很可能直接把數(shù)據(jù)庫打爆。穩(wěn)妥做法Flink的JDBC連接池大小不要超過數(shù)據(jù)庫最大連接數(shù)的20%。第三檢查checkpoint恢復(fù)后的連接狀態(tài)。Flink任務(wù)重啟恢復(fù)時(shí)舊連接可能已經(jīng)失效需要設(shè)置JdbcExecutionOptions的自動(dòng)重連參數(shù)。我在Flink里一般這樣配JdbcExecutionOptions.builder() .withBatchSize(5000) .withBatchIntervalMs(2000) .withMaxRetries(3) .build();4.3 ES寫入報(bào)錯(cuò)與Kafka的InvalidReceiveException日志鏈路的另一個(gè)高頻故障是啟動(dòng)時(shí)ES集群還沒就緒Flink的Sink已經(jīng)開始寫入報(bào)各種節(jié)點(diǎn)不可用、shard lock異常。規(guī)避方法是在鏈路啟動(dòng)前做一次健康檢查——確認(rèn)ES的/_cluster/health返回的statusgreen或者至少yellow再啟動(dòng)Flink作業(yè)。生產(chǎn)環(huán)境建議把健康檢查腳本寫成shell腳本在CI/CD流水線里檢查。還有一個(gè)必須認(rèn)識(shí)清楚的經(jīng)典Kafka報(bào)錯(cuò)org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size 647204481 larger than 100000000)我第一看到這個(gè)報(bào)錯(cuò)時(shí)也懵了后來排查清楚才知道兩個(gè)原因最常見一是客戶端配置的receive.buffer.bytes和broker端不匹配。某次我排查的時(shí)候發(fā)現(xiàn)Flink客戶端的receive.buffer.bytes被設(shè)成了100MB而broker端的socket.request.max.bytes默認(rèn)只有100MB。當(dāng)單條消息大小接近這個(gè)值時(shí)會(huì)觸法這個(gè)異常。解決方案是把message.max.bytes和socket.request.max.bytes在broker端和客戶端都調(diào)大并保持匹配。二是客戶端反序列化框架的版本不一致。Kafka客戶端把字節(jié)流反序列化時(shí)會(huì)校驗(yàn)FrameSize版本不一致或被污染的數(shù)據(jù)流就會(huì)出現(xiàn)這個(gè)異常。排查這種報(bào)錯(cuò)的標(biāo)準(zhǔn)姿勢(shì)是先看客戶端配置再看broker端配置逐個(gè)對(duì)齊同時(shí)檢查兩端Kafka版本是否一致。4.4 Kafka消費(fèi)端多線程如何保證消息順序性日志分析場(chǎng)景對(duì)全局順序的要求通常不高但如果你要處理某個(gè)用戶的完整操作鏈路同一用戶的操作日志必須保證順序。Kafka保證順序性的前提是同一分區(qū)內(nèi)的消息按offset遞增順序消費(fèi)而相同key的消息會(huì)被路由到同一個(gè)分區(qū)。Flink消費(fèi)Kafka時(shí)默認(rèn)以partition為單位做并行消費(fèi)Flink內(nèi)部每個(gè)partition對(duì)應(yīng)一個(gè)subtask天然保證同一個(gè)分區(qū)內(nèi)的消息按順序處理。問題出在多線程處理下游的環(huán)節(jié)如果Flink算子內(nèi)部用了線程池并發(fā)處理消息順序就亂了。我踩過這個(gè)坑后總結(jié)了三個(gè)保序方案按推薦程度排序方案一提高Flink并行度但保證相同key的數(shù)據(jù)進(jìn)同一個(gè)分區(qū)。Flink上游Kafka Source并行度等于Kafka分區(qū)數(shù)保證相同key的消息進(jìn)入同一個(gè)子任務(wù)即可保序。這個(gè)方案最干凈前提是你的并行度要求和分區(qū)數(shù)匹配。方案二關(guān)鍵算子內(nèi)部用單線程。如果你不得不在算子內(nèi)部并發(fā)處理比如外部IO較慢那就要隱藏分區(qū)key讓該key的所有數(shù)據(jù)都被路由到同一個(gè)線程。做法是自定義一個(gè)KeyedProcessFunction內(nèi)部用單線程處理每個(gè)key的數(shù)據(jù)。方案三放棄全局嚴(yán)格順序用事件時(shí)間水位線兜底。日志場(chǎng)景里95%的“順序性問題”其實(shí)可以用窗口和事件時(shí)間優(yōu)雅解決。Flink的Watermark機(jī)制允許一定程度的亂序只要延遲在容忍范圍內(nèi)計(jì)算結(jié)果就是正確的。最后提醒一個(gè)特別容易犯的錯(cuò)誤如果你為了保序而把所有數(shù)據(jù)都發(fā)送到同一個(gè)分區(qū)那你等于放棄了Kafka的并行能力整個(gè)鏈路的吞吐會(huì)驟降。生產(chǎn)環(huán)境優(yōu)先選方案一把保序收斂到key級(jí)別而不是全局。5. 這套方案的后續(xù)擴(kuò)展方向剛才提到的這些都還只是實(shí)時(shí)日志分析的基線能力。鏈路搭好之后往上擴(kuò)展的空間非常大——比如把Flink的Cep模式匹配能力接進(jìn)來做異常行為實(shí)時(shí)告警又比如把日志指標(biāo)輸出到Prometheus用Grafana做基礎(chǔ)監(jiān)控還比如在Flink里接入OpenMetadata自動(dòng)采集Flink作業(yè)的血緣關(guān)系讓數(shù)據(jù)資產(chǎn)的元數(shù)據(jù)跟上實(shí)時(shí)的節(jié)奏。我個(gè)人的體會(huì)是日志分析平臺(tái)永遠(yuǎn)不是靜態(tài)工程它更像一個(gè)不斷生長的基座不停接入新的數(shù)據(jù)源、新的分析維度、新的下游系統(tǒng)。架構(gòu)選型時(shí)如果沒留出擴(kuò)展的余地后面每一次新需求都要傷筋動(dòng)骨。而這套ELKFlinkKafka的組合擴(kuò)展性恰恰是我在工程實(shí)踐中體會(huì)最深的一點(diǎn)。