flink学习续集(小麦)

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: flink

数据准备

这里我们准备了一份淘宝用户行为数据集(来自阿里云天池公开数据集)。本数据集包含了淘宝上某一天随机一百万用户的所有行为(包括点击、购买、加购、收藏)。数据集的组织形式和MovieLens-20M类似,即数据集的每一行表示一条用户行为,由用户ID、商品ID、商品类目ID、行为类型和时间戳组成,并以逗号分隔。关于数据集中每一列的详细描述如下:

列名称 说明
用户ID 整数类型,加密后的用户ID
商品ID 整数类型,加密后的商品ID
商品类目ID 整数类型,加密后的商品所属类目ID
行为类型 字符串,枚举类型,包括(‘pv’, ‘buy’, ‘cart’, ‘fav’)
时间戳 行为发生的时间戳,单位秒

你可以通过下面的命令下载数据集到项目的 resources 目录下:

$ cd my-flink-project/src/main/resources
$ curl https://raw.githubusercontent.com/wuchong/my-flink-project/master/src/main/resources/UserBehavior.csv > UserBehavior.csv

这里是否使用 curl 命令下载数据并不重要,你也可以使用 wget 命令或者直接访问链接下载数据。关键是,将数据文件保存到项目的 resources 目录下,方便应用程序访问。

编写程序

创建模拟数据源

我们先创建一个 UserBehavior 的 POJO 类(所有成员变量声明成public便是POJO类),强类型化后能方便后续的处理。

/**
  * 用户行为数据结构
  **/
public static class UserBehavior {
  public long userId;         // 用户ID
  public long itemId;         // 商品ID
  public int categoryId;      // 商品类目ID
  public String behavior;     // 用户行为, 包括("pv", "buy", "cart", "fav")
  public long timestamp;      // 行为发生的时间戳,单位秒
}

接下来我们就可以创建一个 PojoCsvInputFormat 了, 这是一个读取 csv 文件并将每一行转成指定 POJO
类型(在我们案例中是 UserBehavior)的输入器。

// UserBehavior.csv 的本地文件路径
URL fileUrl = HotItems2.class.getClassLoader().getResource("UserBehavior.csv");
Path filePath = Path.fromLocalFile(new File(fileUrl.toURI()));
// 抽取 UserBehavior 的 TypeInformation,是一个 PojoTypeInfo
PojoTypeInfo<UserBehavior> pojoType = (PojoTypeInfo<UserBehavior>) TypeExtractor.createTypeInfo(UserBehavior.class);
// 由于 Java 反射抽取出的字段顺序是不确定的,需要显式指定下文件中字段的顺序
String[] fieldOrder = new String[]{"userId", "itemId", "categoryId", "behavior", "timestamp"};
// 创建 PojoCsvInputFormat
PojoCsvInputFormat<UserBehavior> csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder);

下一步我们用 PojoCsvInputFormat 创建输入源。

DataStream<UserBehavior> dataSource = env.createInput(csvInput, pojoType);

这就创建了一个 UserBehavior 类型的 DataStream

EventTime与Watermark

当我们说“统计过去一小时内点击量”,这里的“一小时”是指什么呢? 在 Flink 中它可以是指 ProcessingTime ,也可以是 EventTime,由用户决定。

  • ProcessingTime事件被处理的时间。也就是由机器的系统时间来决定
  • EventTime事件发生的时间。一般就是数据本身携带的时间

在本案例中,我们需要统计业务时间上的每小时的点击量,所以要基于 EventTime 来处理。那么如果让 Flink 按照我们想要的业务时间来处理呢?这里主要有两件事情要做:

  • 告诉 Flink 我们现在按照 EventTime 模式进行处理,Flink 默认使用 ProcessingTime 处理,所以我们要显式设置下。

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  • 指定如何获得业务时间,以及生成 Watermark。Watermark 是用来追踪业务事件的概念,可以理解成 EventTime 世界中的时钟,用来指示当前处理到什么时刻的数据了。由于我们的数据源的数据已经经过整理,没有乱序,即事件的时间戳是单调递增的,所以可以将每条数据的业务时间就当做 Watermark。这里我们用 AscendingTimestampExtractor 来实现时间戳的抽取和 Watermark 的生成。

注意:真实业务场景一般都是存在乱序的,所以一般使用 BoundedOutOfOrdernessTimestampExtractor

DataStream<UserBehavior> timedData = dataSource
    .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
      @Override
      public long extractAscendingTimestamp(UserBehavior userBehavior) {
            // 原始数据单位秒,将其转成毫秒
            return userBehavior.timestamp * 1000;
      }
    });

这样我们就得到了一个带有时间标记的数据流了,后面就能做一些窗口的操作。

过滤出点击事件

在开始窗口操作之前,先回顾下需求“每隔5分钟输出过去一小时内点击量最多的前 N 个商品”。由于原始数据中存在点击、加购、购买、收藏各种行为的数据,但是我们只需要统计点击量,所以先使用 FilterFunction 将点击行为数据过滤出来。

DataStream<UserBehavior> pvData = timedData
    .filter(new FilterFunction<UserBehavior>() {
              @Override
              public boolean filter(UserBehavior userBehavior) throws Exception {
                // 过滤出只有点击的数据
                return userBehavior.behavior.equals("pv");
              }
    });

窗口统计点击量

由于要每隔5分钟统计一次最近一小时每个商品的点击量,所以窗口大小是一小时,每隔5分钟滑动一次。即分别要统计 [09:00, 10:00), [09:05, 10:05), [09:10, 10:10)… 等窗口的商品点击量。是一个常见的滑动窗口需求(Sliding Window)。

DataStream<ItemViewCount> windowedData = pvData
     // 对商品进行分组
     .keyBy("itemId")
     // 对每个商品做滑动窗口(1小时窗口,5分钟滑动一次)
      .timeWindow(Time.minutes(60), Time.minutes(5))
     // 做增量的聚合操作,它能使用AggregateFunction提前聚合掉数据,减少 state 的存储压力
     .aggregate(new CountAgg(), new WindowResultFunction());

CountAgg

这里的CountAgg实现了AggregateFunction接口,功能是统计窗口中的条数,即遇到一条数据就加一。

/**
  * COUNT 统计的聚合函数实现,每出现一条记录加一
  **/
public static class CountAgg implements AggregateFunction<UserBehavior, Long, Long> {
      @Override
      public Long createAccumulator() {
              return 0L;
      }

      @Override
      public Long add(UserBehavior userBehavior, Long acc) {
            return acc + 1;
      }

      @Override
      public Long getResult(Long acc) {
            return acc;
      }

      @Override
      public Long merge(Long acc1, Long acc2) {
            return acc1 + acc2;
      }
}

WindowFunction

.aggregate(AggregateFunction af, WindowFunction wf) 的第二个参数WindowFunction将每个 key每个窗口聚合后的结果带上其他信息进行输出。这里实现的WindowResultFunction将主键商品ID,窗口,点击量封装成了ItemViewCount进行输出。

/**
  * 用于输出窗口的结果
  **/
public static class WindowResultFunction implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow> {
      @Override
      public void apply(
              Tuple key,  // 窗口的主键,即 itemId
              TimeWindow window,  // 窗口
              Iterable<Long> aggregateResult, // 聚合函数的结果,即 count 值
              Collector<ItemViewCount> collector  // 输出类型为 ItemViewCount
      ) throws Exception {
            Long itemId = ((Tuple1<Long>) key).f0;
            Long count = aggregateResult.iterator().next();
            collector.collect(ItemViewCount.of(itemId, window.getEnd(), count));
      }
}

/**
  * 商品点击量(窗口操作的输出类型)
  **/
public static class ItemViewCount {
      public long itemId;     // 商品ID
      public long windowEnd;  // 窗口结束时间戳
      public long viewCount;  // 商品的点击量
      public static ItemViewCount of(long itemId, long windowEnd, long viewCount) {
            ItemViewCount result = new ItemViewCount();
            result.itemId = itemId;
            result.windowEnd = windowEnd;
            result.viewCount = viewCount;
            return result;
      }
}

现在我们得到了每个商品在每个窗口的点击量的数据流。

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
6月前
|
SQL 运维 API
Apache Flink 学习教程----持续更新
Apache Flink 学习教程----持续更新
279 0
|
SQL Serverless 程序员
准备数据集用于flink学习
准备一百多万的交易数据,作为flink学习过程中的数据集
114 1
准备数据集用于flink学习
|
流计算
从Flink 重启策略机制能学习到什么?
最近在学习Flink ,在看到Flink的重启策略机制时感觉这个设计很好。
108 0
|
存储 运维 供应链
为什么要学习 Apache Flink| 学习笔记
快速学习为什么要学习 Apache Flink。
379 0
为什么要学习 Apache Flink| 学习笔记
|
资源调度 分布式计算 Hadoop
基于mac构建大数据伪分布式学习环境(十一)-部署Flink1.14.5
本文主要讲解实时计算引擎Flink的部署,并使用word count实例来验证部署结果
158 0
|
SQL 消息中间件 缓存
大数据开发笔记(九):Flink综合学习)(二)
Flink 是一个框架和分布式处理引擎,用于对无界和有界数据流进行有状态计算。并且 Flink 提供了数据分布、容错机制以及资源管理等核心功能。Flink提供了诸多高抽象层的API以便用户编写分布式任务
274 0
大数据开发笔记(九):Flink综合学习)(二)
|
SQL 机器学习/深度学习 存储
大数据开发笔记(九):Flink综合学习)(一)
Flink 是一个框架和分布式处理引擎,用于对无界和有界数据流进行有状态计算。并且 Flink 提供了数据分布、容错机制以及资源管理等核心功能。Flink提供了诸多高抽象层的API以便用户编写分布式任务
199 0
大数据开发笔记(九):Flink综合学习)(一)
|
存储 消息中间件 缓存
学习flink的state
Apache Flink® — Stateful Computations over Data Streams,数据流上的状态计算。可以看出flink默认它是一个默认就有状态的分析引擎,State一般指一个具体的 Task/Operator 的状态,State数据默认保存在 Java 的堆内存中。 假设一个 Task 在处理过程中挂掉了,那么它在内存中的状态都会丢失,所有的数据都需要重新计算。从容错和消息处理的语义(At -least-once 和 Exactly-once)上来说,Flink引入了State 和 CheckPoint。
393 2
学习flink的state
|
存储 缓存 搜索推荐
|
2月前
|
运维 数据处理 数据安全/隐私保护
阿里云实时计算Flink版测评报告
该测评报告详细介绍了阿里云实时计算Flink版在用户行为分析与标签画像中的应用实践,展示了其毫秒级的数据处理能力和高效的开发流程。报告还全面评测了该服务在稳定性、性能、开发运维及安全性方面的卓越表现,并对比自建Flink集群的优势。最后,报告评估了其成本效益,强调了其灵活扩展性和高投资回报率,适合各类实时数据处理需求。