专栏首页最新最全的大数据技术体系Flink应用案例统计实现TopN的两种方式

Flink应用案例统计实现TopN的两种方式

文章目录

Flink应用案例——Top N

窗口的计算处理,在实际应用中非常常见。对于一些比较复杂的需求,如果增量聚合函数 无法满足,我们就需要考虑使用窗口处理函数这样的“大招”了。 网站中一个非常经典的例子,就是实时统计一段时间内的热门 url。例如,需要统计最近 10 秒钟内最热门的两个 url 链接,并且每 5 秒钟更新一次。我们知道,这可以用一个滑动窗口 来实现,而“热门度”一般可以直接用访问量来表示。于是就需要开滑动窗口收集 url 的访问 数据,按照不同的 url 进行统计,而后汇总排序并最终输出前两名。这其实就是著名的“Top N” 问题。 很显然,简单的增量聚合可以得到 url 链接的访问量,但是后续的排序输出 Top N 就很难 实现了。所以接下来我们用窗口处理函数进行实现。

使用 ProcessAllWindowFunction

一种最简单的想法是,我们干脆不区分 url 链接,而是将所有访问数据都收集起来,统一 进行统计计算。所以可以不做 keyBy,直接基于 DataStream 开窗,然后使用全窗口函数 ProcessAllWindowFunction 来进行处理。 在窗口中可以用一个 HashMap 来保存每个 url 的访问次数,只要遍历窗口中的所有数据, 自然就能得到所有 url 的热门度。最后把 HashMap 转成一个列表 ArrayList,然后进行排序、 取出前两名输出就可以了。 代码具体实现如下:

package flink.demo;

import flink.demo.PoJo.Event;
import flink.demo.Source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
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.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
public class ProcessAllWindowTopN {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment
                env
                =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event>
                eventStream
                =
                env.addSource(new ClickSource())
                        .assignTimestampsAndWatermarks(
                                WatermarkStrategy.<Event>forMonotonousTimestamps()
                                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                                                           @Override
                                                                           public long extractTimestamp(Event element, long
                                                                                   recordTimestamp) {
                                                                               return element.timestamp;
                                                                           }
                                                                       })
                        );
// 只需要 url 就可以统计数量,所以转换成 String 直接开窗统计
        SingleOutputStreamOperator<String> result = eventStream
                .map(new MapFunction<Event, String>() {
                    @Override
                    public String map(Event value) throws Exception {
                        return value.url;
                    }
                })
                .windowAll(SlidingEventTimeWindows.of(Time.seconds(10),
                        Time.seconds(5))) // 开滑动窗口
                .process(new ProcessAllWindowFunction<String, String, TimeWindow>()
                {
                    @Override
                    public void process(Context context, Iterable<String> elements,
                                        Collector<String> out) throws Exception {
                        HashMap<String, Long> urlCountMap = new HashMap<>();
// 遍历窗口中数据,将浏览量保存到一个 HashMap 中
                        for (String url : elements) {
                            if (urlCountMap.containsKey(url)) {
                                long count = urlCountMap.get(url);
                                urlCountMap.put(url, count + 1L);
                            } else {
                                urlCountMap.put(url, 1L);
                            }
                        }
                        ArrayList<Tuple2<String,
                                Long>>
                                mapList
                                =
                                new
                                        ArrayList<Tuple2<String, Long>>();
// 将浏览量数据放入 ArrayList,进行排序
                        for (String key : urlCountMap.keySet()) {
                            mapList.add(Tuple2.of(key, urlCountMap.get(key)));
                        }
                        mapList.sort(new Comparator<Tuple2<String, Long>>() {
                            @Override
                            public int compare(Tuple2<String, Long> o1, Tuple2<String,
                                    Long> o2) {
                                return o2.f1.intValue() - o1.f1.intValue();
                            }
                        });
// 取排序后的前两名,构建输出结果
                        StringBuilder result = new StringBuilder();
                        result.append("========================================\n");
                        for (int i = 0; i < 2; i++) {
                            Tuple2<String, Long> temp = mapList.get(i);
                            String info = "浏览量 No." + (i + 1) +
                                    " url:" + temp.f0 +
                                    " 浏览量:" + temp.f1 +
                                    " 窗 口 结 束 时 间:"
                                    +
                                    new
                                            Timestamp(context.window().getEnd()) + "\n";
                            result.append(info);
                        }
                        result.append("========================================\n");
                        out.collect(result.toString());
                    }
                });
        result.print();
        env.execute();
    }
}

运行结果如下所示:

浏览量 No.1 url:./prod?id=1

浏览量:2 窗口结束时间:2021-07-01 15:24:25.0

浏览量 No.2 url:./cart 浏览量:1 窗口结束时间:2021-07-01 15:24:25.0

使用 KeyedProcessFunction

在上一小节的实现过程中,我们没有进行按键分区,直接将所有数据放在一个分区上进行 了开窗操作。这相当于将并行度强行设置为 1,在实际应用中是要尽量避免的,所以 Flink 官 方也并不推荐使用 AllWindowedStream 进行处理。另外,我们在全窗口函数中定义了 HashMap 来统计 url 链接的浏览量,计算过程是要先收集齐所有数据、然后再逐一遍历更新 HashMap, 这显然不够高效。如果我们可以利用增量聚合函数的特性,每来一条数据就更新一次对应 url 的浏览量,那么到窗口触发计算时只需要做排序输出就可以了。 基于这样的想法,我们可以从两个方面去做优化:一是对数据进行按键分区,分别统计浏 览量;二是进行增量聚合,得到结果最后再做排序输出。所以,我们可以使用增量聚合函数 AggregateFunction 进行浏览量的统计,然后结合 ProcessWindowFunction 排序输出来实现 Top N 的需求。 具体实现思路就是,先按照 url 对数据进行 keyBy 分区,然后开窗进行增量聚合。这里就 会发现一个问题:我们进行按键分区之后,窗口的计算就会只针对当前 key 有效了;也就是说, 每个窗口的统计结果中,只会有一个 url 的浏览量,这是无法直接用 ProcessWindowFunction 进行排序的。所以我们只能分成两步:先对每个 url 链接统计出浏览量,然后再将统计结果收 集起来,排序输出最终结果。因为最后的排序还是基于每个时间窗口的,所以为了让输出的统 计结果中包含窗口信息,我们可以借用第六章中定义的 POJO 类 UrlViewCount 来表示,它包 202 含了 url、浏览量(count)以及窗口的起始结束时间。之后对 UrlViewCount 的处理,可以先按 窗口分区,然后用 KeyedProcessFunction 来实现。 总结处理流程如下: (1)读取数据源; (2)筛选浏览行为(pv); (3)提取时间戳并生成水位线; (4)按照 url 进行 keyBy 分区操作; (5)开长度为 1 小时、步长为 5 分钟的事件时间滑动窗口; (6)使用增量聚合函数 AggregateFunction,并结合全窗口函数 WindowFunction 进行窗口 聚合,得到每个 url、在每个统计窗口内的浏览量,包装成 UrlViewCount; (7)按照窗口进行 keyBy 分区操作; (8)对同一窗口的统计结果数据,使用 KeyedProcessFunction 进行收集并排序输出。 糟糕的是,这里又会带来另一个问题。最后我们用 KeyedProcessFunction 来收集数据做排 序,这时面对的就是窗口聚合之后的数据流,而窗口已经不存在了;那到底什么时候会收集齐 所有数据呢?这问题听起来似乎有些没道理。我们统计浏览量的窗口已经关闭,就说明了当前 已经到了要输出结果的时候,直接输出不就行了吗? 没有这么简单。因为数据流中的元素是逐个到来的,所以即使理论上我们应该“同时”收 到很多 url 的浏览量统计结果,实际也是有先后的、只能一条一条处理。下游任务(就是我们 定义的 KeyedProcessFunction)看到一个 url 的统计结果,并不能保证这个时间段的统计数据 不会再来了,所以也不能贸然进行排序输出。解决的办法,自然就是要等所有数据到齐了—— 这很容易让我们联想起水位线设置延迟时间的方法。这里我们也可以“多等一会儿”,等到水 位线真正超过了窗口结束时间,要统计的数据就肯定到齐了。 具体实现上,可以采用一个延迟触发的事件时间定时器。基于窗口的结束时间来设定延迟, 其实并不需要等太久——因为我们是靠水位线的推进来触发定时器,而水位线的含义就是“之 前的数据都到齐了”。所以我们只需要设置 1 毫秒的延迟,就一定可以保证这一点。 而在等待过程中,之前已经到达的数据应该缓存起来,我们这里用一个自定义的“列表状 态”(ListState)来进行存储,如图 7-2 所示。这个状态需要使用富函数类的 getRuntimeContext() 方法获取运行时上下文来定义,我们一般把它放在 open()生命周期方法中。之后每来一个 UrlViewCount,就把它添加到当前的列表状态中,并注册一个触发时间为窗口结束时间加 1 毫秒(windowEnd + 1)的定时器。待到水位线到达这个时间,定时器触发,我们可以保证当 前窗口所有 url 的统计结果 UrlViewCount 都到齐了;于是从状态中取出进行排序输出。

图 7-2 使用“列表状态”进行排序 具体代码实现如下:

import flink.demo.PoJo.Event;
import flink.demo.PoJo.UrlViewCount;
import flink.demo.Source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
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.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
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.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
public class KeyedProcessTopN {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment
            env
            =
            StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
// 从自定义数据源读取数据
    SingleOutputStreamOperator<Event>
            eventStream
            =
            env.addSource(new ClickSource())
                    .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                            .withTimestampAssigner(new
                                                           SerializableTimestampAssigner<Event>() {
                                                             @Override
                                                             public
                                                             long
                                                   extractTimestamp(Event
                                                                                      element,
                                                                              long
                                                                                      recordTimestamp) {
                                                               return element.timestamp;
                                                             }
                                                           }));
// 需要按照 url 分组,求出每个 url 的访问量
    SingleOutputStreamOperator<UrlViewCount> urlCountStream =
            eventStream.keyBy(data -> data.url)
                    .window(SlidingEventTimeWindows.of(Time.seconds(10),
                            Time.seconds(5)))
                    .aggregate(new UrlViewCountAgg(),new UrlViewCountResult());
// 对结果中同一个窗口的统计数据,进行排序处理
    SingleOutputStreamOperator<String> result = urlCountStream.keyBy(data ->
            data.windowEnd)
            .process(new TopN(2));
    result.print("result");
    env.execute();
  }
  // 自定义增量聚合
  public static class UrlViewCountAgg implements AggregateFunction<Event, Long,
  Long> {
    @Override
    public Long createAccumulator() {
      return 0L;
    }
    @Override
    public Long add(Event value, Long accumulator) {
      return accumulator + 1;
    }
    @Override
    public Long getResult(Long accumulator) {
      return accumulator;
    }
    @Override
    public Long merge(Long a, Long b) {
      return null;
    }
  }
  // 自定义全窗口函数,只需要包装窗口信息
  public static class UrlViewCountResult extends ProcessWindowFunction<Long,UrlViewCount, String, TimeWindow> {
    @Override
    public void process(String url, Context context, Iterable<Long> elements,
                        Collector<UrlViewCount> out) throws Exception {
// 结合窗口信息,包装输出内容
      Long start = context.window().getStart();
      Long end = context.window().getEnd();
      out.collect(new UrlViewCount(url, elements.iterator().next(), start,
              end));
    }
  }
  // 自定义处理函数,排序取 top n
  public static class TopN extends KeyedProcessFunction<Long, UrlViewCount,String>{
    // 将 n 作为属性
    private Integer n;
    // 定义一个列表状态
    private ListState<UrlViewCount> urlViewCountListState;
    public TopN(Integer n) {
      this.n = n;
    }
    @Override
    public void open(Configuration parameters) throws Exception {
// 从环境中获取列表状态句柄
      urlViewCountListState = getRuntimeContext().getListState(
              new ListStateDescriptor<UrlViewCount>("url-view-count-list",
                      Types.POJO(UrlViewCount.class)));
    }
    @Override
    public
    void
    processElement(UrlViewCount value,
                   Context
                           ctx,
                   Collector<String> out) throws Exception {
// 将 count 数据添加到列表状态中,保存起来
      urlViewCountListState.add(value);
// 注册 window end + 1ms 后的定时器,等待所有数据到齐开始排序
      ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);
    }
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String>
            out) throws Exception {
// 将数据从列表状态变量中取出,放入 ArrayList,方便排序
      ArrayList<UrlViewCount> urlViewCountArrayList = new ArrayList<>();
      for (UrlViewCount urlViewCount : urlViewCountListState.get()) {
        urlViewCountArrayList.add(urlViewCount);
      }
// 清空状态,释放资源
      urlViewCountListState.clear();
// 排序
      urlViewCountArrayList.sort(new Comparator<UrlViewCount>() {
        @Override
        public int compare(UrlViewCount o1, UrlViewCount o2) {
          return o2.count.intValue() - o1.count.intValue();
        }
      });
// 取前两名,构建输出结果
      StringBuilder result = new StringBuilder();
      result.append("========================================\n");
      result.append("窗口结束时间:" + new Timestamp(timestamp - 1) + "\n");
      for (int i = 0; i < this.n; i++) {
        UrlViewCount UrlViewCount = urlViewCountArrayList.get(i);
        String info = "No." + (i + 1) + " "

                + "url:" + UrlViewCount.url + " "
                + "浏览量:" + UrlViewCount.count + "\n";
        result.append(info);
      }
      result.append("========================================\n");
      out.collect(result.toString());
    }
  }
}

代码中,我们还利用了定时器的特性:针对同一 key、同一时间戳会进行去重。所以对于 同一个窗口而言,我们接到统计结果数据后设定的 windowEnd + 1 的定时器都是一样的,最终 只会触发一次计算。而对于不同的 key(这里 key 是 windowEnd),定时器和状态都是独立的, 所以我们也不用担心不同窗口间数据的干扰。 我们在上面的代码中使用了后面要讲解的 ListState。这里可以先简单说明一下。我们先声 明一个列表状态变量:

private ListState<Event> UrlViewCountListState;

然后在 open 方法中初始化了列表状态变量,我们初始化的时候使用了 ListStateDescriptor 描述符,这个描述符用来告诉 Flink 列表状态变量的名字和类型。列表状态变量是单例,也就 是说只会被实例化一次。这个列表状态变量的作用域是当前 key 所对应的逻辑分区。我们使用 add 方法向列表状态变量中添加数据,使用 get 方法读取列表状态变量中的所有元素。 另外,根据水位线的定义,我们这里的延迟时间设为 0 事实上也是可以保证数据都到齐的。 感兴趣的读者可以自行修改代码进行测试。

本文参与 腾讯云自媒体分享计划 ,欢迎热爱写作的你一起参与!
本文分享自作者个人站点/博客:https://blog.csdn.net/xianyu120复制
如有侵权,请联系 cloudcommunity@tencent.com 删除。
登录 后参与评论
0 条评论

相关文章

  • Flink SQL 知其所以然(二十七):TopN、Order By、Limit 操作

    大家好,我是老羊,今天我们来学习 Flink SQL 中的 TopN、Order By、Limit 3个操作。

    公众号:大数据羊说
  • 实时计算Flink在游戏行业的实时数仓建设实践

    某游戏公司开发了个游戏APP,该公司在APP中会发布一些游戏场景、游戏角色、装备、精美皮肤等内容,玩家在线娱乐,产生充值购买等行为。 业务的构建涉及到几个端:

    用户5877097
  • Flink第一课!使用批处理,流处理,Socket的方式实现经典词频统计

    导入成功之后有一点要注意,就是java_2.12中的2.12指的是scala的版本,导入依赖成功之后即在对应目录创建包与对应类开始项目的编写。

    大数据小禅
  • (中)史上最全干货!Flink SQL 成神之路(全文 18 万字、138 个案例、42 张图)

    CREATE 语句用于向当前或指定的 Catalog 中注册库、表、视图或函数。注册后的库、表、视图和函数可以在 SQL 查询中使用。

    公众号:大数据羊说
  • Flink & 低代码:为应用实时计算铺平道路

    目前京东实时计算平台已经发展到了一定规模,且在 Flink 的应用上也积累了很多经验与反思。本次我们专访了京东数据分析优化部的算法工程师张颖老师,期待能从京东落...

    深度学习与Python
  • ​大数据入门学习指南

    大数据入门核心技术栏目是为初学者精心打造入门大数据必学知识整理,内容十分丰富,集合将近200篇高质文章带你轻松学习。

    Lanson
  • 实战 | flink sql 与微博热搜的碰撞!!!

    根据微博目前站内词条消费情况,计算 top 50 消费热度词条,每分钟更新一次,并且按照列表展现给用户。

    Spark学习技巧
  • 15-Flink实战项目之实时热销排行

    某个图书网站,希望看到双十一秒杀期间实时的热销排行榜单。我们可以将“实时热门商品”翻译成程序员更好理解的需求:每隔5秒钟输出最近一小时内点击量最多的前 N 个商...

    王知无-import_bigdata
  • 基于Flink快速开发实时TopN

    TopN 是统计报表和大屏非常常见的功能,主要用来实时计算排行榜。流式的TopN可以使业务方在内存中按照某个统计指标(如出现次数)计算排名并快速出发出更新后的排...

    王知无-import_bigdata
  • 附代码|Flink实时计算TopN

    在上一章代码中使用了timeWindow,使得我们可以操作Flink流中的一个时间段内的数据,这就引出了Flink中的"窗口"概念:在大多数场景下,数据流都是"...

    小晨说数据
  • 【Flink】小白级入门,Flink sql 的基础用法

    对于用户只需要表达我想要什么,具体处理逻辑交给框架,系统处理,用户无需关心,对于一些非专业的开发人员有了解 sql,并且 sql 相对我们学习 java,c 等...

    857技术社区
  • 基于Flink快速开发实时TopN程序最简单的思路

    TopN 是统计报表和大屏非常常见的功能,主要用来实时计算排行榜。流式的TopN可以使业务方在内存中按照某个统计指标(如出现次数)计算排名并快速出发出更新后的排...

    王知无-import_bigdata
  • Flink企业级优化全面总结(3万字长文,15张图)

    Flink性能调优的第一步,就是为任务分配合适的资源,在一定范围内,增加资源的分配与性能的提升是成正比的,实现了最优的资源配置后,在此基础上再...

    大数据老哥
  • 零基础学Flink:实时热销榜Top5(案例)

    如前文所预告的一样,今天我们来分析一下,如何通过flink完成实时热销榜单Top5的计算,本文案例,需要使用前文一些内容,如果不了解的同学,请移步《零基础学Fl...

    麒思妙想
  • Flink实战: 窗口TopN分析与实现

    TopN 的需求场景不管是在离线计算还是实时计算都是比较常见的,例如电商中计算热门销售商品、广告计算中点击数前N的广告、搜索中计算搜索次数前N的搜索词。topN...

    Flink实战剖析
  • Flink 实践教程:进阶4-窗口 TOP N

    流计算 Oceanus 是大数据产品生态体系的实时化分析利器,是基于 Apache Flink 构建的具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点...

    吴云涛

扫码关注腾讯云开发者

领取腾讯云代金券