Flink海量数据实时去重
Flink海量数据实时去重
方案1: 借助redis的Set
具体实现代码
缺点
- 需要频繁连接Redis
- 如果数据量过大, 对redis的内存也是一种压力
方案2: 使用Flink的MapState
具体实现代码
缺点
- 如果数据量过大, 状态后端最好选择 RocksDBStateBackend
- 如果数据量过大, 对存储也有一定压力
方案3: 使用布隆过滤器
布隆过滤器可以大大减少存储的数据的数据量
优点
- 不需要存储数据本身,只用比特表示,因此空间占用相对于传统方式有巨大的优势,并且能够保密数据;
- 时间效率也较高,插入和查询的时间复杂度均为, 所以他的时间复杂度实际是
- 哈希函数之间相互独立,可以在硬件指令层面并行计算。
缺点
- 存在假阳性的概率,不适用于任何要求100%准确率的情境;
- 只能插入和查询元素,不能删除元素,这与产生假阳性的原因是相同的。我们可以简单地想到通过计数(即将一个比特扩展为计数值)来记录元素数,但仍然无法保证删除的元素一定在集合中。
使用场景
所以,BF在对查准度要求没有那么苛刻,而对时间、空间效率要求较高的场合非常合适.
另外,由于它不存在假阴性问题,所以用作“不存在”逻辑的处理时有奇效,比如可以用来作为缓存系统(如Redis)的缓冲,防止缓存穿透。
使用布隆过滤器实现去重
Flink已经内置了布隆过滤器的实现(使用的是google的Guava)
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();
// 创建WatermarkStrategy
WatermarkStrategy<UserBehavior> wms = WatermarkStrategy
.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
@Override
public long extractTimestamp(UserBehavior element, long recordTimestamp) {
return element.getTimestamp() * 1000L;
}
});
env
.readTextFile("input/UserBehavior.csv")
.map(line -> { // 对数据切割, 然后封装到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())) //过滤出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;
@Override
public 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>>() { })
)
);
}
@Override
public void process(String key,
Context context,
Iterable<UserBehavior> elements, Collector<String> out) throws Exception {
countState.update(0L);
// 在状态中初始化一个布隆过滤器
// 参数1: 漏斗, 存储的类型
// 参数2: 期望插入的元素总个数
// 参数3: 期望的误判率(假阳性率)
BloomFilter<Long> bf = BloomFilter.create(Funnels.longFunnel(), 1000000, 0.001);
bfState.update(bf);
for (UserBehavior behavior : elements) {
// 查布隆
if (!bfState.value().mightContain(behavior.getUserId())) {
// 不存在 计数+1
countState.update(countState.value() + 1L);
// 记录这个用户di, 表示来过
bfState.value().put(behavior.getUserId());
}
}
out.collect("窗口: " + context.window() + " 的uv是: " + countState.value());
}
})
.print();
env.execute();
}
}
还没有评论,来说两句吧...