本培训对Apache Flink进行了自以为是的介绍，包括足够让您开始编写可伸缩的流ETL、分析和事件驱动的应用程序，同时忽略了许多(最终重要的)细节。重点是为Flink的状态和时间管理api提供直观的介绍，希望您掌握了这些基础知识后，能够更好地从文档中了解需要了解的其他内容。
具体来说，你会学到:
- 如何建立一个环境来开发Flink程序
- 如何实现流数据处理管道
- Flink如何以及为什么管理状态
- 如何使用事件时间一致计算准确的分析
- 如何在连续流上构建事件驱动的应用程序
- Flink如何能够提供容错的、有状态的流处理和精确的一次语义

一项目介绍

1.1 基于flink的流处理

首先介绍四个概念：流数据的连续处理（continuous processing of streaming data）、事件时间（event time）、有状态流处理（stateful stream processing）和状态快照（state snapshots）。

批处理是处理有界数据流的范例。在这种操作模式中，我们可以选择在产生任何结果之前摄取整个数据集，这意味着可以对数据进行排序、计算全局统计数据或生成总结所有输入的最终报告。

另一方面，流处理涉及到无限的数据流。至少在概念上，输入可能永远不会结束，因此我们被迫在数据到达时不断地处理它。

在Flink中，应用程序由可以由用户定义的操作符转换的数据流组成。这些数据流形成有向图，以一个或多个源开始，以一个或多个汇聚结束。

应用程序可以使用来自消息队列等流源或来自Apache Kafka或Kinesis等分布式日志的实时数据。但是，flink也可以使用来自各种数据源的有限的历史数据。类似地，由Flink应用程序产生的结果流可以发送到各种各样的系统，并且可以通过REST API访问Flink中的状态。

1.1.1 及时的流处理

对于大多数流媒体应用程序来说，能够使用与处理实时数据相同的代码来重新处理历史数据是非常有价值的，并且无论如何都能产生确定的、一致的结果。

还必须注意事件发生的顺序，而不是它们交付处理的顺序，并能够推断出一组事件何时(或应该何时)完成。例如，考虑电子商务交易或金融交易中涉及的事件集。

通过使用记录在数据流中的事件时间时间戳，而不是使用处理数据的机器的时钟，可以满足及时流处理的这些需求。

1.1.2 有状态的流处理

Flink的操作可以是有状态的。这意味着一个事件的处理方式可以依赖于之前所有事件的累积效果。状态可以用于简单的事情，例如计算仪表板上显示的每分钟事件数，也可以用于更复杂的事情，例如计算欺诈检测模型的功能。

Flink应用程序在分布式集群上并行运行。给定操作符的各种并行实例将在单独的线程中独立执行，并且通常将运行在不同的机器上。

有状态操作符的并行实例集实际上是切分的键值存储。每个并行实例负责处理特定键组的事件，这些键的状态保存在本地。

下图显示了一个作业在作业图的前三个操作符之间以并行度为2的方式运行，直到一个并行度为1的接收为止。第三个操作符是有状态的，我们可以看到在第二个和第三个操作符之间发生了完全连接的网络转移。这样做是为了按某个键对流进行分区，以便将需要一起处理的所有事件都放在一起。

状态总是在本地访问，这有助于Flink应用程序实现高吞吐量和低延迟。您可以选择将状态保存在JVM堆上，或者如果JVM堆太大，可以选择高效组织磁盘上的数据结构。

1.1.3 健壮的流处理

通过状态快照和流回放的组合，Flink能够提供容错的、精确的一次语义。这些快照捕获分布式管道的整个状态，将偏移量记录到输入队列中，并记录在整个作业图中的状态，这是由于在此之前摄入了数据而导致的。当发生故障时，将重新对源进行缠绕，恢复状态并恢复处理。如上所述，这些状态快照是异步捕获的，不会妨碍正在进行的处理。

1.2 什么可以流?

Flink针对Java和Scala的DataStream api将允许您流任何可以序列化的内容。使用Flink自己的序列化器：

基本数据类型：String, Long, Integer, Boolean, Array
复杂数据类型：Tuples, POJOs, and Scala case classes

对于其他类型，Flink会返回到Kryo。

JAVA-Tuple

Tuple2<String, Integer> person = new Tuple2<>("Fred", 35);

// zero based index!

String name = person.f0;

Integer age = person.f1;

1.3 流执行环境

这个分布式运行时取决于您的应用程序是可序列化的。它还要求集群中的每个节点都可以使用所有依赖项。

二实验环境部署

2.1 部署开发环境

2.1.1 软件需求

Java JDK 8

Apache Maven 3.x

Git

IntelliJ（推荐）

2.1.2 复制和构建flink-training-execises 项目

git clone https://github.com/dataArtisans/flink-training-exercises.git

cd flink-training-exercises

mvn clean package

如果您以前没有这样做过，此时您将下载这个Flink训练练习项目的所有依赖项。这通常需要几分钟，取决于你的互联网连接速度。

如果所有测试都通过了，构建成功了，那么您就有了一个良好的开始。

如果在国内，需要在maven的settings.xml中配置一下maven settings镜像：

<id>nexus-aliyun</id>

<name>Nexus aliyun</name>

<url>http://maven.aliyun.com/nexus/content/groups/public</url>

</mirror>

</mirrors>

</settings>

2.1.3 将flink-training-exercises项目导入IDE

因为这个项目混合了Java和Scala代码，你需要安装Scala插件，如果你还没有的话:

进入IntelliJ插件设置(IntelliJ IDEA -> Preferences -> plugins)，点击“安装Jetbrains插件…”。
选择并安装“Scala”插件。
重启IntelliJ

导入项目，选择它的pom.xml文件
在每一步，接受默认值;不选择配置文件
继续，确保在进入SDK对话框时，它有一个到JDK的有效路径，并将所有其他选项保留为默认值，然后完成项目导入
打开项目结构对话框，并在全局库部分添加Scala 2.12 SDK(即使您不打算使用Scala，也需要安装它)

现在应该可以打开以下类全路径。并成功运行此测试。

com.dataartisans.flinktraining.exercises.datastream_java.basic.RideCleansingTest

2.1.4 下载数据集

您可以通过运行以下命令下载本次培训中使用的出租车数据文件数据：

wget http://training.ververica.com/trainingData/nycTaxiRides.gz

wget http://training.ververica.com/trainingData/nycTaxiFares.gz

使用wget或其他方法来获取这些文件并不重要，但是无论如何获取数据，都不要解压或重命名.gz文件。

2.2 使用Taxi数据流

2.2.1 Taxi Ride 事件结构（schema）

我们的出租车数据集包含纽约市出租车的个人信息。每一次骑行都由两个事件表示:旅程开始事件和旅程结束事件。每个项目由11个字段组成:

rideId : Long // a unique id for each ride

taxiId : Long // a unique id for each taxi

driverId : Long // a unique id for each driver

isStart : Boolean // TRUE for ride start events, FALSE for ride end events

startTime : DateTime // the start time of a ride

endTime : DateTime // the end time of a ride, // "1970-01-01 00:00:00" for start events startLon : Float // the longitude of the ride start location

startLat : Float // the latitude of the ride start location

endLon : Float // the longitude of the ride end location

endLat : Float // the latitude of the ride end location

passengerCnt : Short // number of passengers on the ride

注意:数据集包含无效或缺少坐标信息的记录(经度和纬度是0.0)。

还有一个包含出租车车费数据的相关数据集，包括以下字段:

rideId : Long // a unique id for each ride

taxiId : Long // a unique id for each taxi

driverId : Long // a unique id for each driver

startTime : DateTime // the start time of a ride

paymentType : String // CSH or CRD

tip : Float // tip for this ride

tolls : Float // tolls for this ride

totalFare : Float // total fare collected

2.2.2 在Flink程序中生成taxi乘车数据流

我们提供了一个Flink源函数(TaxiRideSource)，它读取一个.gz文件，其中包含出租车出行记录，并发出一个出租车出行事件流。源在事件时操作。对于出租车费用事件，有一个类似的源函数(TaxiFareSource)。

为了尽可能真实地生成流，事件的发出与它们的时间戳成比例。在现实中，两件事发生在十分钟之后，也会在十分钟之后发生。可以指定加速因子来快速前进流，即在美国，如果加速系数为60，那么一分钟内发生的事件会在一秒内送达。此外，可以指定最大服务延迟，这将导致每个事件在指定的范围内随机延迟。这将产生一个无序流，这在许多实际应用程序中很常见。

对于这些练习，加速因子为600或更多(即，每秒钟处理事件时间为10分钟)，最大延迟为60秒。

所有的练习都应该使用event-time特征来实现。事件时间将程序语义与服务速度解耦，并确保即使在历史数据或无序交付的数据的情况下也能得到一致的结果。

Checkpointing

有些练习要求你使用CheckpointedTaxiRideSource和/或CheckpointedTaxiFareSource。与TaxiRideSource和TaxiFareSource不同，这些变量能够检查它们的状态。

Table Sources

还要注意，有TaxiRideTableSource和TaxiFareTableSource表源可用于表和SQL api。

2.3 如何开始实验学习

在动手实践部分，您将使用各种Flink api实现Flink程序。您还将了解如何使用Apache Maven打包Flink程序，并在运行中的Flink实例上执行打包的程序。

以下步骤将指导您完成以下过程:使用提供的数据流、实现第一个Flink流程序，以及在一个正在运行的Flink实例上打包和执行程序。

我们假设您已经根据我们的设置指南设置了您的开发环境，并从github获得了一个本地克隆的flink-train -exercises repo。

2.3.1 修改ExeciseBase数据路径

下载完数据集，打开

com.dataartisans.flinktraining.exercises.datastream_java.utils.ExerciseBase

并编辑下载的数据路径

public final static String pathToRideData = "/Users/david/stuff/flink-training/trainingData/nycTaxiRides.gz";

public final static String pathToFareData = "/Users/david/stuff/flink-training/trainingData/nycTaxiFares.gz";

2.3.2 在IDE中运行和调试Flink程序

在IDE中启动Flink程序与运行它的main()方法一样简单。在底层，执行环境将在同一个进程中启动一个本地Flink实例。因此，也可以在代码中放置断点并调试它。

假设您有一个导入了flink-training-exercises项目的IDE，您可以运行(或调试)一个简单的流作业，如下所示:

打开这个类

com.ververica.flinktraining.examples.datastream_java.basics.RideCount

使用IDE运行(或调试) RideCountExample类的main()方法。

三实验1——清洗数据流

目标：清洗掉开始或者结束经纬度都不在纽约市区域内的行程记录日志。

可以参考GeoUtils. isInNYC(float lon, float lat) 来判断经纬度是否在区域内。

3.1 入口类

Java: com.ververica.flinktraining.exercises.datastream_java.basics.RideCleansingExercise

Scala: com.ververica.flinktraining.exercises.datastream_scala.basics.RideCleansingExercise

3.2 测试类

com.ververica.flinktraining.exercises.datastream_java.basics.RideCleansingTest

四转换数据

4.1 无状态的转换（Stateless Transformations）

Map()

参考类： Enrichment

Flatmap()

参考类：NYCEnrichment

4.2 Keyed Streams

keyBy()

能够围绕某个属性对流进行分区通常非常有用，这样就可以将具有相同属性值的所有事件分组在一起。例如，假设我们想要在每个网格单元中找到最长的出租车旅程。如果我们考虑SQL查询，这意味着使用startCell执行某种GROUP BY，而在Flink中这是使用keyBy(KeySelector)。

每个keyBy都会导致重新划分流的网络shuffle（转移）。通常这是非常昂贵的，因为它涉及到网络通信以及序列化和反序列化。

Aggregations on Keyed Streams

(Implicit) State

reduce() and other aggregators

上面使用的maxBy()只是Flink的KeyedStreams上可用的许多聚合器函数中的一个例子。还有一个更通用的reduce()函数，您可以使用它来实现自己的定制聚合。

4.3 有状态的转换（Stateful Transformations）

4.4 连接流（Connected Streams）

有时，与其应用这样一个预定义的转换，

您希望能够动态地更改转换的某些方面—通过输入阀值、规则或其他参数。Flink中支持这种模式的是所谓的连接流，其中一个操作符有两个输入流，如下所示:

五实验2——Stateful Enrichment

目标：将TaxiRide和TaxiFare记录join一起到每一次ride

5.1 入口类

Java: com.ververica.flinktraining.exercises.datastream_java.state.RidesAndFaresExercise

Scala: com.ververica.flinktraining.exercises.datastream_scala.state.RidesAndFaresExercise

5.2 测试类

com.ververica.flinktraining.exercises.datastream_java.state.RidesAndFaresTest

六时间和分析（Time and Analytics）

6.1 Event Time 和 Watermarks

Flink明确支持三种不同的时间概念:

event time：事件发生的时间，由产生该事件的设备记录；

ingestion time: 当事件被摄入时，由Flink记录的时间戳；

processing time: 管道中的特定操作符处理事件的时间。

Working with Watermarks

DataStream<MyEvent> stream = ...

DataStream<MyEvent> withTimestampsAndWatermarks =

stream.assignTimestampsAndWatermarks(new MyExtractor);

public static class MyExtractor

extends BoundedOutOfOrdernessTimestampExtractor<MyEvent> {

public MyExtractor() {

super(Time.seconds(10));

}

@Override

public long extractTimestamp(MyEvent event) {

return element.getCreationTime();

}

6.2 窗口（Windows）

https://training.ververica.com/lessons/windows.html

七实验3——Windowing

目标：使用窗口时间，“每小时小费”活动任务，找出每小时赚消费最多的司机。

思路：首先，使用一个小时长的窗口，计算每个司机在一小时内的小费总数；

然后，从这个窗口的结果流中找到每个小时消费总数最多的司机。

7.1 入口类

Java: com.ververica.flinktraining.exercises.datastream_java.windows.HourlyTipsExercise

Scala: com.ververica.flinktraining.exercises.datastream_scala.windows.HourlyTipsExercise

7.2 测试类

com.ververica.flinktraining.exercises.datastream_java.windows.HourlyTipsTest

7.3 核心代码

Java代码

DataStream<Tuple3<Long, Long, Float>> hourlyTips = fares

.keyBy((TaxiFare fare) -> fare.driverId)

.timeWindow(Time.hours(1))

.process(new AddTips());

其中ProcessWindowFunction完成所有的繁重工作:

public static class AddTips extends ProcessWindowFunction<

TaxiFare, Tuple3<Long, Long, Float>, Long, TimeWindow> {

@Override

public void process(Long key, Context context, Iterable<TaxiFare> fares, Collector<Tuple3<Long, Long, Float>> out) throws Exception {

Float sumOfTips = 0F;

for (TaxiFare f : fares) {

sumOfTips += f.tip;

}

out.collect(new Tuple3<>(context.window().getEnd(), key, sumOfTips));

}

Scala代码

val hourlyTips = fares

.map((f: TaxiFare) => (f.driverId, f.tip))

.keyBy(_._1)

.timeWindow(Time.hours(1))

.reduce(

(f1: (Long, Float), f2: (Long, Float)) => { (f1._1, f1._2 + f2._2) },

new WrapWithWindowInfo())

class WrapWithWindowInfo() extends ProcessWindowFunction[(Long, Float), (Long, Long, Float), Long, TimeWindow] {

override def process(key: Long, context: Context, elements: Iterable[(Long, Float)], out: Collector[(Long, Long, Float)]): Unit = {

val sumOfTips = elements.iterator.next()._2

out.collect((context.window.getEnd(), key, sumOfTips))

}

八事件驱动应用（Event-driven Apps）

8.1 ProcessFunction

https://training.ververica.com/lessons/processfunction.html

8.2 Side Outputs

九实验4——过期状态（Expiring State）

https://training.ververica.com/exercises/rideEnrichment-processfunction.html

十容错

10.1 State Backends

10.2 Checkpoints 和 Savepoints

https://training.ververica.com/snapshots.html

本文参考：https://training.ververica.com/

flink基础与flink培训——出租车乘车项目相关推荐

【Flink基础】-- Flink CDC介绍
一.Flink CDC 是什么? 2020年 Flink cdc 首次在 Flink forward 大会上官宣, 由 Jark Wu & Qingsheng Ren 两位大佬介绍,原始 b ...
Day74~75_Flink(一）Flink基础核心知识
第一讲 Flink基础核心知识课程大纲课程内容学习效果掌握目标 Flink概述 Flink概述了解 Flink开发环境 Flink开发环境掌握 Flink入门实战 Flink入门实战掌握 ...
Flink 基础原理与优化思路
文章目录什么是Flink? Flink 的特性 Flink 相比传统的 Spark Streaming 有什么区别? Flink 的组件栈有哪些? Flink 的运行必须依赖 Hadoop 组件吗? ...
轻松入门进阶Flink第一课 Flink基础
开篇词:实时计算领域最锋利的武器 Flink 你好,欢迎来到 Flink 专栏,我是王知无,目前在某一线互联网公司从事数据平台架构和研发工作多年,算是整个大数据开发领域的老兵了. 我最早从 Relea ...
学习笔记Flink（四）—— Flink基础API及核心数据结构
一.Flink基础API-Flink编程的基本概念 1.1.Flink程序 Flink 程序是实现了分布式集合转换(例如过滤.映射.更新状态.join.分组.定义窗口.聚合)的规范化程序. 集合初始创 ...
【Flink实时数仓】数据仓库项目实战《四》日志数据分流【DWD】
文章目录 [Flink实时数仓]数据仓库项目实战 <四>日志数据分流-流量域 [DWD] 1.流量域未经加工的事务事实表 1.1主要任务 1.1.1数据清洗(ETL) 1.1.2新老访客状 ...
Flink基础系列20-数据重分区操作
文章目录一.数据重分区参考: 一.数据重分区重分区操作,在DataStream类中可以看到很多Partitioner字眼的类. 其中partitionCustom(-)方法用于自定义重分区. 测 ...
Flink基础系列17-Tranform之多流转换算子
文章目录一.多流转换算子概述 1.1 Split和Select 1.2 Connect和CoMap 1.3 Union 二.代码实现参考: 一.多流转换算子概述多流转换算子一般包括: Split ...
Apache Doris 系列：基础篇-Flink SQL写入Doris
简介本文介绍 Flink SQL如何流式写入 Apache Doris,分为一下几个部分: Flink Doris connector Doris FE 节点配置 Flink SQL 写 Doris ...

flink基础与flink培训——出租车乘车项目

一 项目介绍