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