java Flink（二十一）Flink实战之电商用户访问量统计，统计某个时间段内的访问总量，设计随机key解决数据倾斜问题，解决并行度只是用一个问题。

心已赠人 2022-10-20 13:52 99阅读 0赞

**1、在实际项目中，如果只单纯的统计数量，但是要进行开窗操作，开窗之前是需要keyBy操作的（全窗口除外），因为我们需要统计所有数量，所以keyBy没有意义。**

**所以我们可以设计Tuple将keyBy 的key统一为固定一个****。**

**2、如此操作 所有key固定，并行度是根据key的hashcode取模之后进行重分区，这样key固定 并行度就没有意义。**

### **题目：** ###

**开窗统计一个小时的窗口内总的访问数量，解决数据倾斜问题，使并行度设计更加合理。**

1、首先项目结构

![20210518112931564.png][]

2、文件内容

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70][]

3、pojo类设计

package Bean;
    
    public class UserBehavior {
        //定义私有属性
        private Long userId; //用户Id
        private Long itemId; //分组Id
        private Integer categoryId; //类别Id
        private String behavior; //动作类型
        private Long timestamp; //时间戳
    
        public UserBehavior() {
        }
    
        public UserBehavior(Long userId, Long itemId, Integer categoryId, String behavior, Long timestamp) {
            this.userId = userId;
            this.itemId = itemId;
            this.categoryId = categoryId;
            this.behavior = behavior;
            this.timestamp = timestamp;
        }
    
        public Long getUserId() {
            return userId;
        }
    
        public void setUserId(Long userId) {
            this.userId = userId;
        }
    
        public Long getItemId() {
            return itemId;
        }
    
        public void setItemId(Long itemId) {
            this.itemId = itemId;
        }
    
        public Integer getCategoryId() {
            return categoryId;
        }
    
        public void setCategoryId(Integer categoryId) {
            this.categoryId = categoryId;
        }
    
        public String getBehavior() {
            return behavior;
        }
    
        public void setBehavior(String behavior) {
            this.behavior = behavior;
        }
    
        public Long getTimestamp() {
            return timestamp;
        }
    
        public void setTimestamp(Long timestamp) {
            this.timestamp = timestamp;
        }
    
        @Override
        public String toString() {
            return "UserBehavior{" +
                    "userId=" + userId +
                    ", itemId=" + itemId +
                    ", categoryId=" + categoryId +
                    ", behavior='" + behavior + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }

包装返回类设计：

package Bean;
    
    public class PageViewCount {
        private String url;
        private Long windowEnd;
        private Long count;
    
        public PageViewCount() {
        }
    
        public PageViewCount(String url, Long windowEnd, Long count) {
            this.url = url;
            this.windowEnd = windowEnd;
            this.count = count;
        }
    
        public String getUrl() {
            return url;
        }
    
        public void setUrl(String url) {
            this.url = url;
        }
    
        public Long getWindowEnd() {
            return windowEnd;
        }
    
        public void setWindowEnd(Long windowEnd) {
            this.windowEnd = windowEnd;
        }
    
        public Long getCount() {
            return count;
        }
    
        public void setCount(Long count) {
            this.count = count;
        }
    
        @Override
        public String toString() {
            return "PageViewCount{" +
                    "url='" + url + '\'' +
                    ", windowEnd=" + windowEnd +
                    ", count=" + count +
                    '}';
        }
    }

4、flink程序设计

使用了随机key 解决并行度问题

package Project;
    
    import Bean.ApachLogEvent;
    import Bean.PageViewCount;
    import Bean.UserBehavior;
    import com.sun.xml.internal.ws.assembler.dev.TubelineAssemblyContextUpdater;
    import org.apache.flink.api.common.functions.AggregateFunction;
    import org.apache.flink.api.common.functions.FilterFunction;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.common.state.ValueState;
    import org.apache.flink.api.common.state.ValueStateDescriptor;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
    import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    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 org.apache.flink.util.OutputTag;
    import scala.Int;
    
    import java.net.URL;
    import java.text.SimpleDateFormat;
    import java.util.Random;
    
    public class PageView {
        public static void main(String[] args) throws Exception{
            //创建环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //设置时间语义
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            //设置并行度
            env.setParallelism(4);
    
            URL resource = HotPages.class.getResource("/UserBehavior.csv");
            DataStream<String> inputStream = env.readTextFile(resource.getPath());
            //转换为POJO，分配时间戳和watermark
            DataStream<UserBehavior> dataStream = inputStream
                    .map(new MapFunction<String, UserBehavior>() {
                        public UserBehavior map(String s) throws Exception {
                            String[] fields = s.split(",");
                            return new UserBehavior(new Long(fields[0]), new Long(fields[1]), new Integer(fields[2]), fields[3], new Long(fields[4]));
                        }
                    }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
                        @Override
                        public long extractAscendingTimestamp(UserBehavior element) {
                            return element.getTimestamp()*1000L;
                        }
                    });
    
            //并行任务改进，设计随机key解决数据倾斜问题
            DataStream<PageViewCount> pvStream = dataStream.filter(data -> "pv".equals(data.getBehavior()))
                    .map(new MapFunction<UserBehavior, Tuple2<Integer, Long>>() {
                        @Override
                        public Tuple2<Integer, Long> map(UserBehavior userBehavior) throws Exception {
                            Random random = new Random();
                            return new Tuple2<Integer, Long>(random.nextInt(10), 1L);
                        }
                    })
                    .keyBy(data -> data.f0)
                    .timeWindow(Time.hours(1))
                    .aggregate(new PvCountAgg(), new PvCountResult());
            //将各分区数据汇总起来
            DataStream<PageViewCount> pvResultStream = pvStream.keyBy(PageViewCount::getWindowEnd)
                    //.sum("count");
                    .process(new TotalPvCount());
    
            pvResultStream.print();
            env.execute("pv count job");
        }
    
        //实现自定义预聚合函数
        public static class PvCountAgg implements AggregateFunction<Tuple2<Integer, Long>, Long, Long>{
            @Override
            public Long createAccumulator() {
                return 0L;
            }
    
            @Override
            public Long add(Tuple2<Integer, Long> integerLongTuple2, Long aLong) {
                return aLong+1;
            }
    
            @Override
            public Long getResult(Long aLong) {
                return aLong;
            }
    
            @Override
            public Long merge(Long aLong, Long acc1) {
                return aLong+acc1;
            }
        }
        //实现自定义窗口函数
        public static class PvCountResult implements WindowFunction<Long, PageViewCount, Integer, TimeWindow>{
            @Override
            public void apply(Integer integer, TimeWindow timeWindow, Iterable<Long> iterable, Collector<PageViewCount> collector) throws Exception {
                collector.collect(new PageViewCount(integer.toString(), timeWindow.getEnd(),iterable.iterator().next()));
            }
        }
        //实现自定义处理函数，把相同窗口分组统计的count值叠加
        public static class TotalPvCount extends KeyedProcessFunction<Long, PageViewCount,PageViewCount>{
            //定义状态 保存当前的总count值
            ValueState<Long> totalCountState;
    
            @Override
            public void open(Configuration parameters) throws Exception {
                totalCountState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("total-count",Long.class, 0L));
            }
    
            @Override
            public void processElement(PageViewCount pageViewCount, Context context, Collector<PageViewCount> collector) throws Exception {
                totalCountState.update(totalCountState.value()+pageViewCount.getCount());
                context.timerService().registerProcessingTimeTimer(pageViewCount.getWindowEnd()+1);
            }
    
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<PageViewCount> out) throws Exception {
                //定时器触发 所有count值都到齐 输出当前总count数量
                Long totalCount = totalCountState.value();
                out.collect(new PageViewCount("pv", ctx.getCurrentKey(), totalCount));
                //清空状态
                totalCountState.clear();
            }
        }
    }

5、输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1][]

[20210518112931564.png]: /images/20221020/84b976ade0394bceac56dacfdfc95d96.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70]: /images/20221020/70c98c8fe5ce4fa08112b8c2e4c8f614.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1]: /images/20221020/4cd9e62c28fa44c29920e49b061713a0.png