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