示例工程大數(shù)據(jù)【免費(fèi)下載鏈接】flink-learningflink learning blog. http://www.54tianzhisheng.cn/ 含 Flink 入門、概念、原理、實(shí)戰(zhàn)、性能調(diào)優(yōu)、源碼解析等內(nèi)容。涉及 Flink Connector、Metrics、Library、DataStream API、Table API SQL 等內(nèi)容的學(xué)習(xí)案例還有 Flink 落地應(yīng)用的大型項(xiàng)目案例PVUV、日志存儲(chǔ)、百億數(shù)據(jù)實(shí)時(shí)去重、監(jiān)控告警分享。歡迎大家支持我的專欄《大數(shù)據(jù)實(shí)時(shí)計(jì)算引擎 Flink 實(shí)戰(zhàn)與性能優(yōu)化》項(xiàng)目地址https://gitcode.com/gh_mirrors/fl/flink-learning點(diǎn)擊查看免費(fèi)下載本文內(nèi)容來自 flink-learning 倉(cāng)庫(kù)《Flink 實(shí)戰(zhàn)與性能優(yōu)化》第 12.2 節(jié)《基于 Flink 的百億數(shù)據(jù)去重實(shí)踐》。文章圍繞真實(shí)生產(chǎn)環(huán)境中Kafka 數(shù)據(jù)重復(fù)這一高頻痛點(diǎn)先給出通用去重方案并量化其存儲(chǔ)成本再對(duì)比 BloomFilter、HBase 全局 Set、Flink KeyedState 三條落地路徑最后結(jié)合倉(cāng)庫(kù)中 flink-learning-project-deduplication 模塊的完整源碼講解 KeyedState 去重的實(shí)現(xiàn)細(xì)節(jié)與 RocksDBStateBackend 的調(diào)優(yōu)手段。讀完本文你將掌握百億級(jí)去重的空間成本估算方法、BloomFilter 原理與適用邊界、Flink 狀態(tài)去重的完整編碼范式以及吞吐量調(diào)優(yōu)的關(guān)鍵參數(shù)。1 背景Kafka 中為什么會(huì)出現(xiàn)重復(fù)數(shù)據(jù)典型的 APP 用戶行為日志分析鏈路為手機(jī) APP 端 → Nginx 服務(wù)端 → Logstash/Flume 等 → Kafka → Flink 任務(wù)由于用戶手機(jī)客戶端的網(wǎng)絡(luò)可能出現(xiàn)不穩(wěn)定APP 端上傳日志普遍遵循寧可重復(fù)上報(bào)也不能漏報(bào)日志的策略因此 Kafka 中同一條日志可能出現(xiàn) 2 條或 2 條以上。而 Flink 任務(wù)的數(shù)據(jù)源基本都是 Kafka當(dāng) Kafka 中存在重復(fù)數(shù)據(jù)時(shí)實(shí)時(shí) ETL 或流計(jì)算都必須基于日志主鍵去重否則會(huì)導(dǎo)致計(jì)算結(jié)果偏高。例如用戶 a 在某個(gè)頁(yè)面只點(diǎn)擊了一次但點(diǎn)擊日志在 Kafka 中出現(xiàn) 2 次最終統(tǒng)計(jì)該頁(yè)面點(diǎn)擊數(shù)時(shí)結(jié)果就會(huì)偏高。需要注意這里只列舉了一種造成重復(fù)的可能生產(chǎn)環(huán)境中導(dǎo)致 Kafka 數(shù)據(jù)重復(fù)的因素還有很多例如消費(fèi)端重試、上游重發(fā)、手動(dòng)補(bǔ)數(shù)等。本節(jié)的核心問題是數(shù)據(jù)已經(jīng)重復(fù)了該如何處理。倉(cāng)庫(kù)中與本主題對(duì)應(yīng)的實(shí)戰(zhàn)模塊是 flink-learning-project-deduplication它提供了完整的可運(yùn)行去重案例下文將圍繞它展開。2 去重的通用解決方案全局 Set TTLKafka 數(shù)據(jù)重復(fù)后各種解決方案思路都比較類似維護(hù)一個(gè)全局 Set 集合存放所有已被處理過的主鍵。處理新日志時(shí)將當(dāng)前日志主鍵與歷史 Set 集合比對(duì)若 Set 中已包含當(dāng)前主鍵 → 說明該日志之前已被處理過過濾掉若 Set 中不包含當(dāng)前主鍵 → 正常處理處理完成后將主鍵加入 Set使 Set 永遠(yuǎn)存放所有已被處理過的數(shù)據(jù)。處理流程本身很簡(jiǎn)單關(guān)鍵在于如何維護(hù)這個(gè) Set 集合。以每天 100 億數(shù)據(jù)為規(guī)?；鶞?zhǔn)可以估算其存儲(chǔ)成本主鍵大小由于數(shù)據(jù)量巨大主鍵必須足夠大以保證不沖突。4 字節(jié) int 只能表示約 42 億個(gè)數(shù)每天百億數(shù)據(jù)必然大量沖突會(huì)把不重復(fù)數(shù)據(jù)誤判為重復(fù)。APP 端沒有全局發(fā)號(hào)器通常使用 UUID 作為日志主鍵36 位字符串如f106c4a1-4c6f-41c1-9d30-bbb2b271284a每個(gè)主鍵占 36 字節(jié)。單日空間36 字節(jié) × 100 億 ≈ 360 GB這僅僅是一天必須加 TTL若不加 TTL10 天數(shù)據(jù)占用 3.6 T100 天占用 36 T空間必然爆炸。假設(shè)按天去重、重復(fù)上報(bào)時(shí)間間隔不會(huì)超過 24 小時(shí)則 TTL 可設(shè)為36 小時(shí)。36 小時(shí)窗口內(nèi)的數(shù)據(jù)量100 億 × 1.5 150 億條主鍵。附帶時(shí)間戳每條數(shù)據(jù)帶 TTL 意味著必須額外保存時(shí)間戳如 Redis 中一個(gè) key 設(shè)了 TTL 卻沒有時(shí)間戳就無法判斷何時(shí)清理。主鍵 36 字節(jié) long 時(shí)間戳 8 字節(jié) 每條至少 44 字節(jié)。總空間150 億 × 44 字節(jié) 660 GB。結(jié)論每天百億數(shù)據(jù)量下純 Set 方案至少需要 660 GB 以上存儲(chǔ)空間。這一量化結(jié)果決定了后續(xù)方案選型的走向要么用低成本的近似結(jié)構(gòu)BloomFilter要么把 Set 放到廉價(jià)的分布式存儲(chǔ)HBase要么借助 Flink 自身的分布式狀態(tài)KeyedState RocksDB天然分?jǐn)偞鎯?chǔ)。3 方案一BloomFilter 實(shí)現(xiàn)去重有些流計(jì)算場(chǎng)景對(duì)準(zhǔn)確性要求并不高例如傳統(tǒng) Lambda 架構(gòu)中會(huì)有離線結(jié)果矯正實(shí)時(shí)結(jié)果。當(dāng)業(yè)務(wù)可以接受小量誤差時(shí)可以使用低成本數(shù)據(jù)結(jié)構(gòu)。BloomFilter 與 HyperLogLog是兩類典型代表且都存在誤差HyperLogLog 只能估算插入了多少個(gè)不重復(fù)元素不能回答是否插入過某個(gè)元素BloomFilter 恰好相反它能告訴你肯定不包含元素 a或可能包含元素 b但不能告訴你里面插入了多少個(gè)元素。3.1 從 bitmap 位圖說起假設(shè)有 1 千萬(wàn)個(gè)整數(shù)數(shù)據(jù)范圍 0 ~ 2000 萬(wàn)如何快速判斷某個(gè)整數(shù)是否在其中方案 AHashMap1000 萬(wàn)個(gè) int 約1000 萬(wàn) × 4 字節(jié) ≈ 40 MB。方案 Bboolean 數(shù)組申請(qǐng)長(zhǎng)度 2000 萬(wàn)的 boolean 數(shù)組以整數(shù)為下標(biāo)值為 true 表示存在。查詢整數(shù) K 時(shí)直接取array[K]。Java 的 boolean 占 1 字節(jié)需要 2000 萬(wàn)字節(jié)。方案 C二進(jìn)制位圖用二進(jìn)制位模擬布爾類型1 表示 true、0 表示 false只需 2000 萬(wàn) bit ≈2.4 MB是 boolean 數(shù)組的 1/8相比 40 MB 原始數(shù)據(jù)方案更是大幅縮減。但如果數(shù)據(jù)范圍擴(kuò)大到 0 ~ 100 億位圖就需要 100 億 bit ≈ 1200 MB反而比存原始數(shù)據(jù)還大。此時(shí)可以壓縮位圖長(zhǎng)度 hash 映射只申請(qǐng) 1 億 bit對(duì) 1000 萬(wàn)個(gè)數(shù)求 hash 映射到位上約 12 MB但會(huì)引入hash 沖突。例如 3 與 100000003 對(duì) 1 億求余都為 3兩者映射到同一個(gè) bit如果集合中包含 100000003位圖下標(biāo) 3 為 1查詢 3 時(shí)就會(huì)誤判存在。為了減少?zèng)_突誕生了 BloomFilter。3.2 BloomFilter 原理hash 沖突無法完全避免于是 BloomFilter 引入多個(gè) hash 函數(shù)只要任意一個(gè)hash 函數(shù)發(fā)現(xiàn)元素不在集合中則該元素肯定不在集合中只有當(dāng)所有hash 函數(shù)都命中了才認(rèn)為元素可能存在于集合中。插入過程以 3 個(gè) hash 函數(shù)為例插入元素 a 時(shí)3 個(gè)函數(shù)分別算出位下標(biāo) 2、8、10將這 3 個(gè)位置 1插入元素 b 時(shí)算出 5、10、14 并置 1下標(biāo) 10 被 a、b 共同涉及。查找過程查元素 c 時(shí)3 個(gè)函數(shù)算出 2、6、9其中 6、9 為 0說明 c肯定不在查元素 d 時(shí)算出 5、8、14三者均為 1于是判定 d可能存在但實(shí)際上 d 從未插入——這正是因?yàn)?a、b 恰好覆蓋了這 3 個(gè)位產(chǎn)生了誤判。由此得到 BloomFilter 的兩個(gè)核心性質(zhì)不包含的結(jié)論是絕對(duì)可靠的包含的結(jié)論存在一定誤判率。誤判率與位數(shù)組長(zhǎng)度、hash 函數(shù)個(gè)數(shù)、已插入元素?cái)?shù)量相關(guān)插入元素越少誤判率越低。3.3 使用 Redis BloomFilter 去重Redis 4.0 之后 BloomFilter 以插件形式加入 Redis。創(chuàng)建時(shí)支持設(shè)定預(yù)期容量預(yù)計(jì)插入的數(shù)據(jù)量與誤判率插入量達(dá)到預(yù)期容量時(shí)的誤判概率。經(jīng)筆者測(cè)試申請(qǐng)預(yù)期容量 10 億、誤判率千分之一的 BloomFilter約需143 億 bit ≈ 14 GB相比 660 GB 的精準(zhǔn) Set 方案存儲(chǔ)成本大幅下降。使用中需注意記錄 BloomFilter 中已插入的元素個(gè)數(shù)當(dāng)插入量達(dá)到預(yù)期容量10 億時(shí)為了保障誤判率應(yīng)清除當(dāng)前 BloomFilter 并重新申請(qǐng)一個(gè)新的。適用邊界BloomFilter 有誤差只能用于能承受一定誤差的場(chǎng)景如日志分析、UV 類近似統(tǒng)計(jì)對(duì)于廣告計(jì)費(fèi)等對(duì)數(shù)據(jù)精度要求極高的場(chǎng)景應(yīng)使用精準(zhǔn)去重方案。4 方案二HBase 維護(hù)全局 Set 實(shí)現(xiàn)去重回顧第 2 節(jié)的估算百億精準(zhǔn)去重需要維護(hù) 150 億條主鍵的 Set每條 44 字節(jié)共需660 GB 存儲(chǔ)空間。注意這里說的是存儲(chǔ)空間而非內(nèi)存空間——660 G 內(nèi)存太貴660 G 的 Redis 云服務(wù)一個(gè)月至少 2 萬(wàn) RMB 以上設(shè)計(jì)架構(gòu)必須考慮成本。HBase 基于 RowKey 的 Get 效率很高因此可以將這個(gè)大 Set 以HBase RowKey形式存儲(chǔ)HBase 表設(shè)置TTL 為 36 小時(shí)最近 36 小時(shí)的 150 億條日志主鍵全部以 RowKey 形式存放每來一條數(shù)據(jù)先拿主鍵去 HBase Get 查詢存在 → 已處理過過濾不存在 → 正常處理處理完成后將主鍵 Put 進(jìn) HBase 表由于數(shù)據(jù)量巨大必須提前對(duì) HBase 表做預(yù)分區(qū)將讀寫壓力分散到各個(gè) RegionServer。4.1 HBase RowKey 去重帶來的問題從工程實(shí)踐角度分析該方案主要面臨以下挑戰(zhàn)本節(jié)在原文基礎(chǔ)上補(bǔ)充說明預(yù)分區(qū)策略要求高RowKey 是 UUID 類隨機(jī)字符串分布雖均勻但預(yù)分區(qū)數(shù)量需按 150 億量級(jí)估算分區(qū)過少會(huì)造成單個(gè) Region 數(shù)據(jù)膨脹與熱點(diǎn)分區(qū)過多則管理成本上升讀放大與寫放大每條日志至少一次 Get 一次 Put百億規(guī)模下對(duì) HBase 集群的 RPC 壓力、WAL 寫入壓力都很大需要足夠的 RegionServer 與合理的表/列族參數(shù)支撐成本與運(yùn)維雖然 HBase 存儲(chǔ)比同容量 Redis 內(nèi)存便宜很多但仍需單獨(dú)維護(hù)一套 HBase 集群或復(fù)用已有集群且去重邏輯依賴外部存儲(chǔ)鏈路變長(zhǎng)、延遲增加。因此是否選擇 HBase 方案需要在額外維護(hù)分布式存儲(chǔ)與換用 Flink 自身狀態(tài)之間權(quán)衡——后者的優(yōu)勢(shì)恰恰在于不需要引入任何外部存儲(chǔ)。5 方案三核心使用 Flink KeyedState 實(shí)現(xiàn)去重倉(cāng)庫(kù)中該方案對(duì)應(yīng) KeyedStateDeduplication.java是本節(jié)重點(diǎn)。5.1 使用 Flink 狀態(tài)維護(hù) Set 的優(yōu)勢(shì)相比 Redis/HBase 外部 SetFlink 狀態(tài)方案的核心優(yōu)勢(shì)無需額外存儲(chǔ)組件Set 直接作為 Flink 算子狀態(tài)存在省去 Redis/HBase 集群的部署與成本天然分布式分?jǐn)侹eyedState 按 key 分布在各個(gè)并行子任務(wù)上百億主鍵的存儲(chǔ)壓力被 TaskManager 集群分?jǐn)偨柚?RocksDB 狀態(tài)后端落盤使用 RocksDBStateBackend 時(shí)狀態(tài)存儲(chǔ)在本地磁盤配合內(nèi)存緩存成本遠(yuǎn)低于純內(nèi)存并支持增量 Checkpoint快照成本可控原生 TTL 支持Flink 的StateTtlConfig可以直接給狀態(tài)設(shè)置 36 小時(shí)過期時(shí)間過期數(shù)據(jù)自動(dòng)清理正好對(duì)應(yīng)第 2 節(jié)必須加 TTL的核心設(shè)計(jì)Checkpoint 容錯(cuò)狀態(tài)隨 Checkpoint 持久化作業(yè)故障恢復(fù)后去重記錄不丟失。5.2 如何使用 KeyedState 維護(hù) Set 集合整體思路對(duì)日志主鍵做 keyBy每個(gè) key 對(duì)應(yīng)一個(gè)ValueStateBoolean標(biāo)記該主鍵是否已處理過。第一次出現(xiàn)時(shí)狀態(tài)為 null處理后update(true)再次出現(xiàn)時(shí)狀態(tài)非 null直接過濾。完整代碼見 KeyedStateDeduplication.java拆解如下。① 環(huán)境與狀態(tài)后端配置主方法前段StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(6); // 使用 RocksDBStateBackend 做為狀態(tài)后端并開啟增量 Checkpoint RocksDBStateBackend rocksDBStateBackend new RocksDBStateBackend( hdfs:///flink/checkpoints, true); rocksDBStateBackend.setNumberOfTransferingThreads(3); // 設(shè)置為機(jī)械硬盤內(nèi)存模式強(qiáng)烈建議為 RocksDB 配備 SSD rocksDBStateBackend.setPredefinedOptions( PredefinedOptions.SPINNING_DISK_OPTIMIZED_HIGH_MEM); env.setStateBackend(rocksDBStateBackend); // Checkpoint 間隔為 10 分鐘 env.enableCheckpointing(TimeUnit.MINUTES.toMillis(10)); // 配置 Checkpoint CheckpointConfig checkpointConf env.getCheckpointConfig(); checkpointConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); checkpointConf.setMinPauseBetweenCheckpoints(TimeUnit.MINUTES.toMillis(8)); checkpointConf.setCheckpointTimeout(TimeUnit.MINUTES.toMillis(20)); checkpointConf.enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);要點(diǎn)狀態(tài)后端指向hdfs:///flink/checkpoints并開啟增量 CheckpointsetNumberOfTransferingThreads(3)控制狀態(tài)恢復(fù)/快照時(shí)的線程數(shù)預(yù)定義模式選擇機(jī)械硬盤高內(nèi)存的組合SPINNING_DISK_OPTIMIZED_HIGH_MEM。② Kafka Consumer 與主鍵 keyByProperties props new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, DeduplicationExampleUtil.broker_list); props.put(ConsumerConfig.GROUP_ID_CONFIG, keyed-state-deduplication); FlinkKafkaConsumerBaseString kafkaConsumer new FlinkKafkaConsumer( DeduplicationExampleUtil.topic, new SimpleStringSchema(), props) .setStartFromGroupOffsets(); env.addSource(kafkaConsumer) .map(log - GsonUtil.fromJson(log, UserVisitWebEvent.class)) // 反序列化 JSON .keyBy((KeySelectorUserVisitWebEvent, String) UserVisitWebEvent::getId) .addSink(new KeyedStateSink());即以日志主鍵idUUID 字符串作為 key 進(jìn)行keyBy。③ KeyedStateSinkValueState TTL 36 小時(shí)去重核心邏輯public static class KeyedStateSink extends RichSinkFunctionUserVisitWebEvent { // 使用該 ValueState 來標(biāo)識(shí)當(dāng)前 Key 是否之前存在過 private ValueStateBoolean isExist; Override public void open(Configuration parameters) throws Exception { super.open(parameters); ValueStateDescriptorBoolean keyedStateDuplicated new ValueStateDescriptor(KeyedStateDeduplication, TypeInformation.of(new TypeHintBoolean() { })); // 狀態(tài) TTL 相關(guān)配置過期時(shí)間設(shè)定為 36 小時(shí) StateTtlConfig ttlConfig StateTtlConfig .newBuilder(Time.hours(36)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility( StateTtlConfig.StateVisibility.NeverReturnExpired) .cleanupInRocksdbCompactFilter(50000000L) .build(); // 開啟 TTL keyedStateDuplicated.enableTimeToLive(ttlConfig); // 從狀態(tài)后端恢復(fù)狀態(tài) isExist getRuntimeContext().getState(keyedStateDuplicated); } Override public void invoke(UserVisitWebEvent value, Context context) throws Exception { // 當(dāng)前 key 第一次出現(xiàn)時(shí)isExist.value() 會(huì)返回 null if (null isExist.value()) { // ... 這里執(zhí)行代碼處理的邏輯 // 執(zhí)行完處理邏輯后更新狀態(tài)值 isExist.update(true); } else { // isExist.value() 不為 null表示當(dāng)前 key 之前已被處理過當(dāng)前數(shù)據(jù)應(yīng)被過濾 } } }幾個(gè)關(guān)鍵設(shè)計(jì)狀態(tài)類型ValueStateBoolean每個(gè) key 只存 1 個(gè)布爾值配合 RocksDB 落盤空間成本極低遠(yuǎn)小于 44 字節(jié)/條的通用 Set 估算TTL 配置過期時(shí)間 36 小時(shí)UpdateType.OnCreateAndWrite表示只在創(chuàng)建和寫入時(shí)更新過期時(shí)間讀取不刷新避免持續(xù)活躍的主鍵永不失效StateVisibility.NeverReturnExpired保證已過期狀態(tài)不會(huì)被讀到cleanupInRocksdbCompactFilter(50000000L)表示在 RocksDB compaction 時(shí)對(duì)過期狀態(tài)進(jìn)行清理每處理 5000 萬(wàn)條狀態(tài)條目觸發(fā)一次避免 TTL 過期數(shù)據(jù)長(zhǎng)期占用磁盤判斷邏輯isExist.value() null表示 key 第一次出現(xiàn) → 執(zhí)行正常處理邏輯并update(true)非 null 表示重復(fù) → 過濾。5.3 數(shù)據(jù)模型與模擬數(shù)據(jù)生成去重的日志模型為 UserVisitWebEvent.java包含id日志唯一 id、date日期如 20191025、pageId頁(yè)面 id、userId用戶唯一標(biāo)識(shí)、url頁(yè)面 url五個(gè)字段。模擬數(shù)據(jù)生成器為 DeduplicationExampleUtil.java核心邏輯public static final String broker_list 192.168.30.215:9092,192.168.30.216:9092,192.168.30.220:9092; public static final String topic user-visit-log-topic; // 每個(gè)事件用 UUID 生成唯一 id其余字段隨機(jī)生成 UserVisitWebEvent userVisitWebEvent UserVisitWebEvent.builder() .id(UUID.randomUUID().toString()) // 日志的唯一 id .date(yyyyMMdd) // 日期 .pageId(pageId) // 頁(yè)面 id .userId(Integer.toString(userId)) // 用戶 id .url(url/ pageId) // 頁(yè)面的 url .build(); ProducerRecord record new ProducerRecordString, String(topic, null, null, GsonUtil.toJson(userVisitWebEvent)); producer.send(record);main方法中循環(huán)Thread.sleep(100)writeToKafka()即每 100ms 向 Kafka topicuser-visit-log-topic寫入一批 JSON 日志。Flink 任務(wù)消費(fèi)同一 topicKafka broker 地址與 topic 需與 KeyedStateDeduplication.java 中保持一致。5.4 優(yōu)化主鍵hash 成 long減少狀態(tài)大小并提高吞吐量keyBy使用 36 位 UUID 字符串key 本身較長(zhǎng)。倉(cāng)庫(kù)提供了優(yōu)化版 TuningKeyedStateDeduplication.java思路是先用 murmur3_128 把 UUID 字符串 hash 成 long再以 long 作為 keyBy 的 keyenv.addSource(kafkaConsumer) .map(string - GsonUtil.fromJson(string, UserVisitWebEvent.class)) // 反序列化 JSON // 這里將日志的主鍵 id 通過 murmur3_128 hash 后將生成 long 類型數(shù)據(jù)當(dāng)做 key .keyBy((KeySelectorUserVisitWebEvent, Long) log - Hashing.murmur3_128(5).hashUnencodedChars(log.getId()).asLong()) .addSink(new KeyedStateDeduplication.KeyedStateSink());該優(yōu)化帶來的收益可以從實(shí)現(xiàn)推斷減少 key 序列化與比較成本long 固定 8 字節(jié)遠(yuǎn)小于 36 字符的 UUID 字符串key 的序列化開銷、網(wǎng)絡(luò) shuffle 開銷與狀態(tài)內(nèi)部比較開銷都更小有利于提升吞吐量key 分布更均勻murmur3_128 是高質(zhì)量哈希能把 UUID 字符串均勻散列到 long 空間配合并行度 6 可使各并行子任務(wù)負(fù)載更均衡去重語(yǔ)義不變對(duì)相同字符串 hash 結(jié)果恒定同一條日志的主鍵仍會(huì)落到同一個(gè) key 上ValueState去重邏輯完全復(fù)用KeyedStateDeduplication.KeyedStateSink。注意hash 理論上存在極小概率的碰撞此優(yōu)化適用于可容忍該概率的業(yè)務(wù)與 BloomFilter 的取舍類似對(duì)精度要求嚴(yán)格的場(chǎng)景建議直接用字符串主鍵。另外優(yōu)化版通過.setStartFromLatest()從最新 offset 消費(fèi)便于聯(lián)調(diào)觀察。6 使用 RocksDBStateBackend 的優(yōu)化方法運(yùn)行上述方案時(shí)如果出現(xiàn)吞吐量時(shí)高時(shí)低、或?qū)崪y(cè)吞吐量偏低的情況可以從以下幾個(gè)方面調(diào)優(yōu)對(duì)應(yīng)原文 12.2.5 節(jié)以下結(jié)合倉(cāng)庫(kù)源碼展開。6.1 設(shè)置本地 RocksDB 的數(shù)據(jù)目錄RocksDBStateBackend 把狀態(tài)存到 TaskManager 本地磁盤RocksDB 的本地?cái)?shù)據(jù)目錄可通過 Flink 配置項(xiàng)state.backend.rocksdb.local-dir指定。百億去重場(chǎng)景狀態(tài)量大、讀寫頻繁建議將本地目錄掛載到SSD源碼注釋也明確建議強(qiáng)烈建議為 RocksDB 配備 SSD配置多塊磁盤目錄逗號(hào)分隔讓 RocksDB 在多塊盤間分?jǐn)?IO緩解單盤瓶頸避免與系統(tǒng)盤、日志盤共用目錄防止 IO 互相干擾。6.2 Checkpoint 參數(shù)相關(guān)配置KeyedStateDeduplication.java 中給出了完整可用的 Checkpoint 配置參數(shù)含義如下參數(shù)代碼配置說明Checkpoint 間隔enableCheckpointing(10 min)每隔 10 分鐘觸發(fā)一次 Checkpoint間隔越大快照頻率越低、對(duì)吞吐影響越小但故障恢復(fù)丟失的數(shù)據(jù)窗口越大Checkpoint 模式EXACTLY_ONCE精確一次語(yǔ)義配合去重保證結(jié)果不重不丟兩次 Checkpoint 最小間隔setMinPauseBetweenCheckpoints(8 min)防止頻繁觸發(fā) Checkpoint給業(yè)務(wù)處理留出穩(wěn)定時(shí)間片Checkpoint 超時(shí)setCheckpointTimeout(20 min)超過 20 分鐘未完成則視為失敗防止快照卡住拖垮作業(yè)外部 CheckpointRETAIN_ON_CANCELLATION作業(yè)取消后保留 Checkpoint便于手動(dòng)恢復(fù)或遷移增量 Checkpoint 已通過new RocksDBStateBackend(checkpointPath, true)的第二個(gè)參數(shù)開啟只上傳新增/變更的 state 文件大幅減少百億狀態(tài)下的快照傳輸量。6.3 RocksDB 參數(shù)相關(guān)配置倉(cāng)庫(kù)代碼中的 RocksDB 相關(guān)配置主要有兩處預(yù)定義優(yōu)化模式setPredefinedOptions(PredefinedOptions.SPINNING_DISK_OPTIMIZED_HIGH_MEM)。SPINNING_DISK_OPTIMIZED_HIGH_MEM是機(jī)械硬盤 高內(nèi)存組合——增加 block cache / write buffer 等內(nèi)存占用以換取磁盤 IO 減少適合狀態(tài)讀寫密集的去重場(chǎng)景若集群配備 SSD 且內(nèi)存有限可評(píng)估換用 SSD 類預(yù)定義選項(xiàng)。狀態(tài)遷移線程數(shù)setNumberOfTransferingThreads(3)控制 Checkpoint 上傳/恢復(fù)時(shí)并行傳輸狀態(tài)文件的線程數(shù)適當(dāng)調(diào)大可縮短 Checkpoint 時(shí)間。TTL 清理策略StateTtlConfig中的cleanupInRocksdbCompactFilter(50000000L)在 RocksDB compaction 階段異步清理過期狀態(tài)避免 36 小時(shí) TTL 的過期主鍵長(zhǎng)期占用磁盤相關(guān)配置位于 KeyedStateDeduplication.java。調(diào)優(yōu)建議吞吐量波動(dòng)時(shí)優(yōu)先檢查 RocksDB 所在磁盤 IO是否機(jī)械盤、是否與 HDFS 寫路徑爭(zhēng)搶 IO、Checkpoint 是否拖慢主流程觀察minPauseBetweenCheckpoints與實(shí)際完成時(shí)間再逐步調(diào)整 block cache、write buffer 等 RocksDB 內(nèi)存參數(shù)。7 小結(jié)與反思本文圍繞百億數(shù)據(jù)去重依次對(duì)比了四條技術(shù)路線方案存儲(chǔ)成本百億/天量級(jí)準(zhǔn)確性適用場(chǎng)景通用全局 SetRedis 等≥ 660 GB內(nèi)存成本極高精準(zhǔn)數(shù)據(jù)量小或預(yù)算充足BloomFilter約 14 GB10 億容量、千分之一誤判率有誤差可容忍誤差的統(tǒng)計(jì)場(chǎng)景HBase 全局 Set660 GB 級(jí)廉價(jià)存儲(chǔ) 集群運(yùn)維精準(zhǔn)已有 HBase 集群、接受鏈路延遲Flink KeyedState RocksDB狀態(tài)落盤、成本低精準(zhǔn)流計(jì)算內(nèi)部去重推薦優(yōu)先考慮核心設(shè)計(jì)原則貫穿全文去重必須帶 TTL36 小時(shí) TTL 使 Set 規(guī)模穩(wěn)定在 150 億條量級(jí)否則空間必然爆炸按精度需求選型能接受誤差選 BloomFilter追求精確選 KeyedState或 HBase善用 Flink 原生能力KeyedState StateTtlConfig RocksDBStateBackend增量 Checkpoint 預(yù)定義優(yōu)化選項(xiàng)即可在流內(nèi)部低成本完成百億級(jí)精準(zhǔn)去重?zé)o需引入外部存儲(chǔ)性能調(diào)優(yōu)從存儲(chǔ)介質(zhì)開始RocksDB 強(qiáng)烈建議配 SSD同時(shí)關(guān)注 Checkpoint 頻率與 RocksDB 內(nèi)存/IO 參數(shù)。如需進(jìn)一步閱讀與運(yùn)行文章原文books/flink-in-action-12.2.md《Flink 實(shí)戰(zhàn)與性能優(yōu)化》第 12.2 節(jié)去重實(shí)戰(zhàn)模塊flink-learning-project-deduplication精準(zhǔn)去重實(shí)現(xiàn)KeyedStateDeduplication.java主鍵 hash 優(yōu)化實(shí)現(xiàn)TuningKeyedStateDeduplication.java日志模型與模擬數(shù)據(jù)UserVisitWebEvent.java、DeduplicationExampleUtil.java運(yùn)行方式將模擬數(shù)據(jù)生成器與 Flink 任務(wù)部署到可訪問 Kafka 的環(huán)境broker 地址見DeduplicationExampleUtil.broker_listtopic 為user-visit-log-topic先運(yùn)行DeduplicationExampleUtil.main持續(xù)造數(shù)再啟動(dòng)KeyedStateDeduplication.main或TuningKeyedStateDeduplication.main觀察去重效果即可。贊分享示例工程大數(shù)據(jù)【免費(fèi)下載鏈接】flink-learningflink learning blog. http://www.54tianzhisheng.cn/ 含 Flink 入門、概念、原理、實(shí)戰(zhàn)、性能調(diào)優(yōu)、源碼解析等內(nèi)容。涉及 Flink Connector、Metrics、Library、DataStream API、Table API SQL 等內(nèi)容的學(xué)習(xí)案例還有 Flink 落地應(yīng)用的大型項(xiàng)目案例PVUV、日志存儲(chǔ)、百億數(shù)據(jù)實(shí)時(shí)去重、監(jiān)控告警分享。歡迎大家支持我的專欄《大數(shù)據(jù)實(shí)時(shí)計(jì)算引擎 Flink 實(shí)戰(zhàn)與性能優(yōu)化》項(xiàng)目地址https://gitcode.com/gh_mirrors/fl/flink-learning點(diǎn)擊查看免費(fèi)下載相關(guān)推薦基于 Apache Flink 的百億數(shù)據(jù)去重實(shí)踐KeyedState TTL 與 RocksDB 調(diào)優(yōu)flink-learning 項(xiàng)目實(shí)戰(zhàn)基于 Apache Flink 的百億數(shù)據(jù)去重實(shí)踐KeyedState TTL 與 RocksDB 調(diào)優(yōu)flink learning 項(xiàng)目實(shí)戰(zhàn) 本篇文示例工程大數(shù)據(jù)百億數(shù)據(jù)實(shí)時(shí)去重Flink狀態(tài)后端選型與優(yōu)化百億數(shù)據(jù)實(shí)時(shí)去重Flink狀態(tài)后端選型與優(yōu)化 本文深入探討了在百億級(jí)別數(shù)據(jù)量下實(shí)現(xiàn)實(shí)時(shí)去重的技術(shù)挑戰(zhàn)與解決方案。文章首先分析了用戶行為日志分析、電商交易訂單去示例工程大數(shù)據(jù)Flink SQL 去重Deduplication基于 ROW_NUMBER() 的流式去重寫法與優(yōu)化器實(shí)現(xiàn)Flink SQL 去重Deduplication基于 ROW_NUMBER 的流式去重寫法與優(yōu)化器實(shí)現(xiàn) 本文以 Flink 官方文檔《Deduplica后端大數(shù)據(jù)流處理批處理創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考