Flink海量数据实时去重

男娘i 2022-11-07 14:56 216阅读 0赞

# Flink海量数据实时去重 #

## 方案1: 借助redis的Set ##

具体实现代码

**缺点**

1.  需要频繁连接Redis
2.  如果数据量过大, 对redis的内存也是一种压力

## 方案2: 使用Flink的MapState ##

具体实现代码

**缺点**

1.  如果数据量过大, 状态后端最好选择 RocksDBStateBackend
2.  如果数据量过大, 对存储也有一定压力

## 方案3: 使用布隆过滤器 ##

布隆过滤器可以大大减少存储的数据的数据量

**优点**

1.  不需要存储数据本身，只用比特表示，因此空间占用相对于传统方式有巨大的优势，并且能够保密数据；
2.  时间效率也较高，插入和查询的时间复杂度均为, 所以他的时间复杂度实际是
3.  哈希函数之间相互独立，可以在硬件指令层面并行计算。

**缺点**

1.  存在假阳性的概率，不适用于任何要求100%准确率的情境；
2.  只能插入和查询元素，不能删除元素，这与产生假阳性的原因是相同的。我们可以简单地想到通过计数（即将一个比特扩展为计数值）来记录元素数，但仍然无法保证删除的元素一定在集合中。

**使用场景**

所以，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();
        }
    }