一、海量數(shù)據(jù)實(shí)時(shí)去重說(shuō)明

借助redis的Set，需要頻繁連接Redis，如果數(shù)據(jù)量過(guò)大, 對(duì)redis的內(nèi)存也是一種壓力；使用Flink的MapState，如果數(shù)據(jù)量過(guò)大, 狀態(tài)后端最好選擇 RocksDBStateBackend； 使用布隆過(guò)濾器，布隆過(guò)濾器可以大大減少存儲(chǔ)的數(shù)據(jù)的數(shù)據(jù)量。

二、海里書(shū)實(shí)時(shí)去重為什么需要布隆過(guò)濾器

如果想判斷一個(gè)元素是不是在一個(gè)集合里，一般想到的是將集合中所有元素保存起來(lái)，然后通過(guò)比較確定。鏈表、樹(shù)、散列表（又叫哈希表，Hash table）等等數(shù)據(jù)結(jié)構(gòu)都是這種思路。
但是隨著集合中元素的增加，我們需要的存儲(chǔ)空間越來(lái)越大。同時(shí)檢索速度也越來(lái)越慢，上述三種結(jié)構(gòu)的檢索時(shí)間復(fù)雜度分別為。
布隆過(guò)濾器即可以解決存儲(chǔ)空間的問(wèn)題, 又可以解決時(shí)間復(fù)雜度的問(wèn)題.
布隆過(guò)濾器的原理是，當(dāng)一個(gè)元素被加入集合時(shí)，通過(guò)K個(gè)散列函數(shù)將這個(gè)元素映射成一個(gè)位數(shù)組中的K個(gè)點(diǎn)，把它們置為1。檢索時(shí)，我們只要看看這些點(diǎn)是不是都是1就（大約）知道集合中有沒(méi)有它了：如果這些點(diǎn)有任何一個(gè)0，則被檢元素一定不在；如果都是1，則被檢元素很可能在。這就是布隆過(guò)濾器的基本思想。

三、布隆過(guò)濾基本概念

布隆過(guò)濾器（Bloom Filter，下文簡(jiǎn)稱(chēng)BF）由Burton Howard Bloom在1970年提出，是一種空間效率高的概率型數(shù)據(jù)結(jié)構(gòu)。它專(zhuān)門(mén)用來(lái)檢測(cè)集合中是否存在特定的元素。
它實(shí)際上是一個(gè)很長(zhǎng)的二進(jìn)制向量和一系列隨機(jī)映射函數(shù)。

實(shí)現(xiàn)原理
布隆過(guò)濾器的原理是，當(dāng)一個(gè)元素被加入集合時(shí)，通過(guò)K個(gè)散列函數(shù)將這個(gè)元素映射成一個(gè)位數(shù)組中的K個(gè)點(diǎn)，把它們置為1。檢索時(shí)，我們只要看看這些點(diǎn)是不是都是1就（大約）知道集合中有沒(méi)有它了：如果這些點(diǎn)有任何一個(gè)0，則被檢元素一定不在；如果都是1，則被檢元素很可能在。這就是布隆過(guò)濾器的基本思想。
BF是由一個(gè)長(zhǎng)度為m比特的位數(shù)組（bit array）與k個(gè)哈希函數(shù)（hash function）組成的數(shù)據(jù)結(jié)構(gòu)。位數(shù)組均初始化為0，所有哈希函數(shù)都可以分別把輸入數(shù)據(jù)盡量均勻地散列。
當(dāng)要插入一個(gè)元素時(shí)，將其數(shù)據(jù)分別輸入k個(gè)哈希函數(shù)，產(chǎn)生k個(gè)哈希值。以哈希值作為位數(shù)組中的下標(biāo)，將所有k個(gè)對(duì)應(yīng)的比特置為1。
當(dāng)要查詢(xún)（即判斷是否存在）一個(gè)元素時(shí)，同樣將其數(shù)據(jù)輸入哈希函數(shù)，然后檢查對(duì)應(yīng)的k個(gè)比特。如果有任意一個(gè)比特為0，表明該元素一定不在集合中。如果所有比特均為1，表明該集合有（較大的）可能性在集合中。為什么不是一定在集合中呢？因?yàn)橐粋€(gè)比特被置為1有可能會(huì)受到其他元素的影響(hash碰撞)，這就是所謂“假陽(yáng)性”（false positive）。相對(duì)地，“假陰性”（false negative）在BF中是絕不會(huì)出現(xiàn)的。
下圖示出一個(gè)m=18, k=3的BF示例。集合中的x、y、z三個(gè)元素通過(guò)3個(gè)不同的哈希函數(shù)散列到位數(shù)組中。當(dāng)查詢(xún)?cè)豾時(shí)，因?yàn)橛幸粋€(gè)比特為0，因此w不在該集合中。

優(yōu)點(diǎn)
1.不需要存儲(chǔ)數(shù)據(jù)本身，只用比特表示，因此空間占用相對(duì)于傳統(tǒng)方式有巨大的優(yōu)勢(shì)，并且能夠保密數(shù)據(jù)；
2.時(shí)間效率也較高，插入和查詢(xún)的時(shí)間復(fù)雜度均為, 所以他的時(shí)間復(fù)雜度實(shí)際是
3.哈希函數(shù)之間相互獨(dú)立，可以在硬件指令層面并行計(jì)算。
缺點(diǎn)
1.存在假陽(yáng)性的概率，不適用于任何要求100%準(zhǔn)確率的情境；
2.只能插入和查詢(xún)?cè)?#xff0c;不能刪除元素，這與產(chǎn)生假陽(yáng)性的原因是相同的。我們可以簡(jiǎn)單地想到通過(guò)計(jì)數(shù)（即將一個(gè)比特?cái)U(kuò)展為計(jì)數(shù)值）來(lái)記錄元素?cái)?shù)，但仍然無(wú)法保證刪除的元素一定在集合中。
使用場(chǎng)景
所以，BF在對(duì)查準(zhǔn)度要求沒(méi)有那么苛刻，而對(duì)時(shí)間、空間效率要求較高的場(chǎng)合非常合適.
另外，由于它不存在假陰性問(wèn)題，所以用作“不存在”邏輯的處理時(shí)有奇效，比如可以用來(lái)作為緩存系統(tǒng)（如Redis）的緩沖，防止緩存穿透。
假陽(yáng)性概率的計(jì)算
假陽(yáng)性的概率其實(shí)就是一個(gè)不在的元素，被k個(gè)函數(shù)函數(shù)散列到的k個(gè)位置全部都是1的概率?？梢园凑杖缦碌牟襟E進(jìn)行計(jì)算： p = f(m,n,k)
其中各個(gè)字母的含義:
1.n ：放入BF中的元素的總個(gè)數(shù)；
2.m：BF的總長(zhǎng)度，也就是bit數(shù)組的個(gè)數(shù)
3.k：哈希函數(shù)的個(gè)數(shù)；
4.p：表示BF將一個(gè)不在其中的元素錯(cuò)判為在其中的概率，也就是false positive的概率；
A.BF中的任何一個(gè)bit在第一個(gè)元素的第一個(gè)hash函數(shù)執(zhí)行完之后為 0的概率是：

B.BF中的任何一個(gè)bit在第一個(gè)元素的k個(gè)hash函數(shù)執(zhí)行完之后為 0的概率是：

C.BF中的任何一個(gè)bit在所有的n元素都添加完之后為 0的概率是:

D.BF中的任何一個(gè)bit在所有的n元素都添加完之后為 1的概率是:

E.一個(gè)不存在的元素被k個(gè)hash函數(shù)映射后k個(gè)bit都是1的概率是：

結(jié)論:在哈數(shù)函數(shù)個(gè)數(shù)k一定的情況下
1.比特?cái)?shù)組m長(zhǎng)度越大, p越小, 表示假陽(yáng)性率越低
2.已插入的元素個(gè)數(shù)n越大, p越大, 表示假陽(yáng)性率越大
經(jīng)過(guò)各種數(shù)學(xué)推導(dǎo):
對(duì)于給定的m和n，使得假陽(yáng)性率（誤判率）最小的k通過(guò)如下公式定義：

四、使用布隆過(guò)濾器實(shí)現(xiàn)去重

Flink已經(jīng)內(nèi)置了布隆過(guò)濾器的實(shí)現(xiàn)(使用的是google的Guava)

package com.lyh.flink12;import com.atguigu.flink.java.chapter_6.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.time.Duration;public class Flink02_UV_BoomFilter {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 創(chuàng)建WatermarkStrategyWatermarkStrategy<UserBehavior> wms = WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {@Overridepublic long extractTimestamp(UserBehavior element, long recordTimestamp) {return element.getTimestamp() * 1000L;}});env.readTextFile("input/UserBehavior.csv").map(line -> { // 對(duì)數(shù)據(jù)切割, 然后封裝到POJO中String[] split = line.split(",");return new UserBehavior(Long.valueOf(split[0]), Long.valueOf(split[1]), Integer.valueOf(split[2]), split[3], Long.valueOf(split[4]));}).filter(behavior -> "pv".equals(behavior.getBehavior())) //過(guò)濾出pv行為.assignTimestampsAndWatermarks(wms).keyBy(UserBehavior::getBehavior).window(TumblingEventTimeWindows.of(Time.minutes(60))).process(new ProcessWindowFunction<UserBehavior, String, String, TimeWindow>() {private ValueState<Long> countState;private ValueState<BloomFilter<Long>> bfState;@Overridepublic void open(Configuration parameters) throws Exception {countState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("countState", Long.class));bfState = getRuntimeContext().getState(new ValueStateDescriptor<BloomFilter<Long>>("bfState", TypeInformation.of(new TypeHint<BloomFilter<Long>>() {})));}@Overridepublic void process(String key,Context context,Iterable<UserBehavior> elements, Collector<String> out) throws Exception {countState.update(0L);// 在狀態(tài)中初始化一個(gè)布隆過(guò)濾器// 參數(shù)1: 漏斗, 存儲(chǔ)的類(lèi)型// 參數(shù)2: 期望插入的元素總個(gè)數(shù)// 參數(shù)3: 期望的誤判率(假陽(yáng)性率)BloomFilter<Long> bf = BloomFilter.create(Funnels.longFunnel(), 1000000, 0.001);bfState.update(bf);for (UserBehavior behavior : elements) {// 查布隆if (!bfState.value().mightContain(behavior.getUserId())) {// 不存在 計(jì)數(shù)+1countState.update(countState.value() + 1L);// 記錄這個(gè)用戶(hù)di, 表示來(lái)過(guò)bfState.value().put(behavior.getUserId());}}out.collect("窗口: " + context.window() + " 的uv是: " + countState.value());}}).print();env.execute();}
}