六、Flink 数据流
Flink数据流
在Flink中，应用程序由数据流组成，这些数据流可以由用户定义的运算符（注：有时我们称这些运算符为“算子”）进行转换。这些数据流形成有向图，从一个或多个源开始，以一个或多个输出结束。



Flink支持流处理和批处理，它是一个分布式的流批结合的大数据处理引擎。在Flink中，认为所有的数据本质上都是随时间产生的流数据，把批数据看作是流数据的特例，只不过流数据是一个无界的数据流，而批数据是一个有界的数据流（例如固定大小的数据集）。如下图所示：



因此，Flink是一个用于在无界和有界数据流上进行有状态计算的通用的处理框架，它既具有处理无界流的复杂功能，也具有专门的运算符来高效地处理有界流。通常我们称无界数据为实时数据，来自消息队列或分布式日志等流源（如Apache Kafka或Kinesis）。而有界数据，通常指的是历史数据，来自各种数据源（如文件、关系型数据库等）。由Flink应用程序产生的结果流可以发送到各种各样的系统，并且可以通过REST API访问Flink中包含的状态。



当Flink处理一个有界的数据流时，就是采用的批处理工作模式。在这种操作模式中，我们可以选择先读取整个数据集，然后对数据进行排序、计算全局统计数据或生成总结所有输入的最终报告。 当Flink处理一个无界的数据流时，就是采用的流处理工作模式。对于流数据处理，输入可能永远不会结束，因此我们必须在数据到达时持续不断地对这些数据进行处理。

Flink分层API
Flink提供了开发流/批处理应用程序的不同抽象层次。如下图所示：



Flink提供了三个分层的API。每个API在简洁性和表达性之间提供了不同的权衡，并针对不同的应用场景。



七.Flink 算子
Flink既可以进行批处理（DataSet），也可以进行实时处理（DataStream）。

将Flink的算子分为两大类：一类是DataSet，一类是DataStream。

Apache Flink 的 DataSet API 提供了一系列的转换操作（Transformations）和动作操作（Operations），这些操作可以用来处理批数据集（DataSets）。以下是一些常用的 DataSet 批处理算子及其在 Scala 中的使用示例。

map - 对数据集中的每个元素应用一个函数，并返回一个新的数据集。

val input: DataSet[String] = ...
val output: DataSet[Int] = input.map(_.length)
flatMap - 类似于 map，但是可以返回多个元素。

val input: DataSet[String] = ...
val output: DataSet[String] = input.flatMap(_.split(" "))
filter - 根据条件过滤数据集中的元素。

val input: DataSet[Int] = ...
val output: DataSet[Int] = input.filter(_ % 2 == 0)
reduce - 对数据集中的元素进行累加操作。

val input: DataSet[Int] = ...
val result: Int = input.reduce(_ + _)
fold - 类似于 reduce，但是提供了一个初始值。

val input: DataSet[Int] = ...
val initial: Int = 0
val result: Int = input.fold(initial)(_ + _)
groupBy - 根据给定的键将数据集分组。

val input: DataSet[(String, Int)] = ...
val keyed: GroupedDataSet[(String, Int)] = input.groupBy(_._1)
coGroup - 将两个数据集按照指定的键进行分组并联合。

val input1: DataSet[(String, Int)] = ...
val input2: DataSet[(String, Int)] = ...
val coGrouped: CoGroupedDataSet[(String, Int), (String, Int)] = input1.coGroup(input2)
join - 根据给定的键将两个数据集进行连接。

val input1: DataSet[(String, Int)] = ...
val input2: DataSet[(String, Int)] = ...
val joined: DataSet[(String, (Int, Int))] = input1.join(input2).where(_._1).equalTo(_._1).map(_._1 -> _._2._1)
window - 对数据集进行窗口操作。

val input: DataSet[(String, Int)] = ...
val windowed: WindowedDataStream[(String, Int), TimeWindow] = input.window(TumblingEventTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
aggregate - 对数据集进行聚合操作。

val input: DataSet[(String, Int)] = ...
val aggregated: DataSet[(String, Int)] = input.aggregate(new MyAggregateFunction)
iterate - 对数据集进行迭代操作。

val input: DataSet[Int] = ...
val iterated: DataSet[Int] = input.iterate(10, i => i < 100)
broadcast - 将数据集广播到其他数据集的每个元素。

val broadcastSet: DataSet[String] = ...
val input: DataSet[Int] = ...
val result: DataSet[(String, Int)] = input.broadcast(broadcastSet).flatMap((i, b) => b.split(" ").map((_, i)))
请注意，上述代码示例仅用于展示算子的基本用法，实际使用时可能需要根据具体业务逻辑进行调整。此外，Flink 还提供了其他的算子和功能，如自定义函数、状态管理等，以支持更复杂的数据处理需求。在使用这些算子时，需要确保正确导入相关的 Flink 库和类。

八.读取各种数据源
在 Apache Flink 中，使用 DataStream API 可以从各种数据源获取数据。以下是一些常见的数据源以及如何在 Scala 中使用它们来创建 DataStream 的示例和描述。

1. 文件数据源
文件数据源是最常见的数据源之一，可以从文件中读取数据。Flink 支持读取文本文件、CSV 文件等。

readTextFile(String path)：逐行读取路径指定的文本文件，即符合TextInputFormat规范的文本文件，并以字符串形式返回。
readFile(FileInputFormat inputFormat, String path)：根据指定的文件输入格式读取(一次)文件。
readFile(fileInputFormat, path, watchType, interval, pathFilter)：这是前两个方法在内部调用的方法。它根据给定的fileInputFormat读取路径中的文件。
// 创建执行环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment()

// 从文件读取数据，例如 /path/to/file.txt
val fileStream: DataStream[String] = env.readTextFile("/path/to/file.txt")

// 打印数据流
fileStream.print()

// 启动 Flink 作业
env.execute("Flink DataStream File Source Example")
2. Socket 数据源
Socket 数据源允许 Flink 从 TCP 套接字读取数据。

socketTextStream(hostName, port)
socketTextStream(hostName, port, delimiter)：可指定分隔符。
socketTextStream(hostName, port, delimiter, maxRetry)：还可以指定API应该尝试获取数据的最大次数。
// 创建执行环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment()

// 从 Socket 读取数据，例如监听 localhost 的 9999 端口
val socketStream: DataStream[String] = env.socketTextStream("localhost", 9999)

// 打印数据流
socketStream.print()

// 启动 Flink 作业
env.execute("Flink DataStream Socket Source Example")
3. 集合数据源
集合数据源允许你直接从 Scala 集合创建一个 DataStream。

fromCollection(Seq)：从Java Java.util. collection创建一个数据流。集合中的所有元素必须具有相同的类型。
fromCollection(Iterator)：从迭代器创建数据流。该类指定迭代器返回的元素的数据类型。
fromElements(elements: _*)：根据给定的对象序列创建数据流。所有对象必须具有相同的类型。
fromParallelCollection(SplittableIterator)：并行地从迭代器创建数据流。该类指定迭代器返回的元素的数据类型。
generateSequence(from, to)：并行地生成给定区间内的数字序列。
// 创建执行环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment()

// 从 Scala 集合创建 DataStream
val collectionStream: DataStream[String] = env.fromCollection(Seq("Flink", "DataStream", "Example"))

// 打印数据流
collectionStream.print()

// 启动 Flink 作业
env.execute("Flink DataStream Collection Source Example")
4. Kafka 数据源
Flink 可以连接到 Kafka 并从 Kafka 主题读取数据。

import org.apache.flink.streaming.connectors.kafka.{KafkaSource, KafkaDeserializationSchema}
import org.apache.flink.api.common.serialization.SimpleStringSchema

// 创建执行环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment()

// 定义 Kafka 连接配置
val kafkaProps = new Properties()
kafkaProps.setProperty("bootstrap.servers", "localhost:9092")
kafkaProps.setProperty("group.id", "flink-kafka-example")

// 创建 Kafka 数据源
val kafkaStream: DataStream[String] = env
.addSource(new KafkaSource[String](kafkaProps, "your-topic", new SimpleStringSchema()))

// 打印数据流
kafkaStream.print()

// 启动 Flink 作业
env.execute("Flink DataStream Kafka Source Example")

5. 自定义数据源
Flink 允许你通过实现 SourceFunction 接口来创建自定义数据源。

import org.apache.flink.streaming.api.functions.source.SourceFunction

// 定义一个自定义数据源
class CustomSource extends SourceFunction[String] {
private var running = true

override def run(sourceContext: SourceContext[String]): Unit = {
for (i <- 1 to 10) {
sourceContext.collect("Element: " + i)
Thread.sleep(1000) // 模拟数据生成延迟
}
running = false
}

override def cancel(): Unit = running = false
}

// 创建执行环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment()

// 添加自定义数据源
val customStream: DataStream[String] = env.addSource(new CustomSource)

// 打印数据流
customStream.print()

// 启动 Flink 作业
env.execute("Flink DataStream Custom Source Example")

以上示例展示了如何在 Scala 中使用 Flink DataStream API 来从不同的数据源获取数据。每个示例都包含了创建数据流的代码和对数据流进行操作的注释。这些示例可以作为构建更复杂 Flink 流处理作业的起点。

九.流处理中的Time与Window
时间是流应用程序的另一个重要组成部分。大多数事件流都具有固有的时间语义，因为每个事件都在特定的时间点生成。此外，许多常见的流计算都是基于时间的，比如窗口聚合、会话、模式检测和基于时间的连接。

Flink提供了一组丰富的与时间相关的特性：

事件时间模式：流应用程序（使用事件时间语义处理流）基于事件的时间戳计算结果。因此，事件时间处理允许精确和一致的结果。
处理时间模式：除了事件时间模式，Flink还支持处理时间语义，它执行由处理机器的挂钟时间触发的计算。处理时间模式可以适用于某些具有严格低延迟要求的应用程序，这些应用程序可以容忍近似结果。
水印支持：Flink在事件时间应用程序中使用水印来推断时间。水印是一种灵活的机制，用来平衡结果的延迟和完整性。
迟到数据处理：当以事件时间模式处理带有水印的流时，可能会在所有相关事件到达之前完成计算。这样的事件称为迟到事件。Flink提供了多个处理迟到事件的选项，比如通过侧输出重新路由它们，并更新以前完成的结果。
在 Flink 的流处理中，时间（Time）和窗口（Window）是两个核心概念，它们共同为处理无界数据流提供了强大的机制。

时间概念
Flink Streaming API借鉴了谷歌数据流模型，它的流API明确支持三个不同的时间概念：

事件时间：事件发生的时间，由产生(或存储)事件的设备记录。
接入时间：Flink在接入事件时记录的时间戳。
处理时间：管道中特定操作符处理事件的时间。
设置时间特性
时间特性定义了系统如何为依赖时间的顺序和依赖时间的操作（如时间窗口）确定时间。默认情况下，Flink DataStream程序将使用EventTime（事件时间）。如果要改用处理时间，那么需要在一开始就设置时间特性。

// 获得流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment

// 设置流的时间特性（这里设置为采用处理时间）
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
注：在Flink 1.12之前，Flink DataStream默认使用的是处理时间。从Flink 1.12开始，默认的流时间特性已被更改为EventTime，因此不再需要调用此方法来启用事件时间支持。

当然也可以选择设置其他类型时间特性。

env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
事件时间和水印
支持事件时间的流处理器需要一种方法来度量事件时间的进度。例如，针对事件时间对数据进行窗口或排序的操作符必须缓冲数据，直到它们能够确保已接收到某个时间间隔的所有时间戳为止。这是由所谓的“时间水印”来处理的。

在Flink中测量事件时间进展的机制是水印（watermarks），水印是一种特殊类型的事件，是告诉系统事件时间进度的一种方式。水印流是数据流的一部分，并带有时间戳t。水印(t)声明事件时间已经到达该流中的时间t，这意味着时间戳t ' <= t（即时间戳更早或等于水印的事件）的流中不应该有更多的元素。



时间t的水印标记了数据流中的一个位置，并断言此时的流在时间t之前已经完成。为了执行基于事件时间的事件处理，Flink需要知道与每个事件相关联的时间，它还需要包含水印的流。水印就是系统事件时间的时钟。水印触发基于事件时间的计时器的触发。

下图显示了带有（逻辑的）时间戳的事件流，以及内联流动的水印。在这个例子中，事件是按顺序排列的（相对于它们的时间戳），这意味着水印只是流中的周期标记。



对于无序流，水印是至关重要的，如下图所示，其中事件不是按照它们的时间戳排序的。



例如，当操作符接收到w(11)这条水印时，可以认为时间戳小于或等于11的数据已经到达，此时可以触发计算。同样，当接收到w(17)这条水印时，可以认为时间戳小于等于17的数据已经到达，此时可以触发计算。

可以看出，水印的时间戳是单调递增的，时间戳为t的水印意味着所有后续记录的时间戳将大于t。一般来说，水印是一种声明，在流中的那个点之前，在某个时间戳之前的所有事件都应该已经到达。当水印到达运算符（算子）时，运算符可以将其内部事件时间时钟推进到水印的值。

时间（Time）
Flink 支持三种时间类型，它们分别是：

事件时间（Event Time）：这是数据事件发生的实际时间，通常由数据源中的时间戳表示。事件时间是处理乱序事件和确保一致性的关键。为了使用事件时间，需要定义水印（Watermarks）来表示事件时间的进度。

摄取时间（Ingestion Time）：这是数据进入 Flink 系统的时间。摄取时间不依赖于事件的实际发生时间，而是依赖于数据到达 Flink 系统的时间。摄取时间通常用于快速处理数据，但不适合处理乱序数据。

处理时间（Processing Time）：这是 Flink 任务执行操作的时间。处理时间与系统时钟相关，适用于不需要事件时间一致性的实时处理场景。

窗口（Window）
窗口是流处理中的一种机制，用于将无界数据流划分为有界的片段，以便于进行聚合和其他计算。Flink 提供了多种窗口类型：

时间窗口（Time Window）：根据时间将数据分组到窗口中。时间窗口可以是翻滚的（Tumbling）或滑动的（Sliding）。

翻滚时间窗口（Tumbling Time Window）：将时间轴分割成固定大小的不重叠窗口。每个数据元素只能属于一个窗口。
滑动时间窗口（Sliding Time Window）：与翻滚窗口类似，但窗口有重叠。一个数据元素可以属于多个窗口。
计数窗口（Count Window）：根据数据元素的数量将数据分组到窗口中。当窗口中的元素数量达到预设的阈值时，窗口会被触发处理。

会话窗口（Session Window）：根据数据中的活动间隙来分组。会话窗口可以动态地根据数据的活跃度来打开和关闭，适用于用户交互等场景。

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

// 创建执行环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

// 定义带有事件时间戳的数据流
val stream: DataStream[(String, Int, Long)] = ???

// 使用事件时间戳和水印定义时间特征
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
stream.assignTimestampsAndWatermarks(WatermarkStrategy.noWatermarks())

// 应用翻滚时间窗口进行处理
val tumblingWindowedStream: DataStream[(String, Int)] = stream
.keyBy(_._1)
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.sum(2) // 假设我们对每个窗口中的整数进行求和

// 应用滑动时间窗口进行处理
val slidingWindowedStream: DataStream[(String, Int)] = stream
.keyBy(_._1)
.window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1)))
.sum(2) // 假设我们对每个窗口中的整数进行求和

// 打印结果并执行作业
tumblingWindowedStream.print()
slidingWindowedStream.print()
env.execute("Flink Time and Window Example")

十.处理函数
在 Flink 中，ProcessFunction 是一个强大的处理函数，它允许用户对流中的每个元素进行复杂的处理，包括状态管理和定时器设置。ProcessFunction 提供了更细粒度的控制，适用于需要维护状态或实现事件时间处理的复杂逻辑。

以下是 ProcessFunction 的一些主要特点和作用：

状态管理 - ProcessFunction 允许用户创建和操作状态，这使得可以跟踪元素的历史信息或执行基于状态的决策。
定时器 - 用户可以在 ProcessFunction 中设置事件时间或处理时间的定时器，以便在将来的某个时间点接收通知。
延迟处理 - ProcessFunction 可以处理延迟数据，即在事件发生后一段时间内到达的数据。
复杂的业务逻辑 - 由于其灵活性，ProcessFunction 可以用来实现复杂的业务逻辑，如窗口聚合、事件模式匹配等。
MapFunction

作用：对每个数据流中的元素应用一个一对一的转换逻辑。
import org.apache.flink.api.common.functions.MapFunction

class MultiplyByTwoMap extends MapFunction[Int, Int] {
override def map(value: Int): Int = {
// 将传入的整数值翻倍
value * 2
}
}

val env = StreamExecutionEnvironment.getExecutionEnvironment
val inputDataStream = env.fromElements(1, 2, 3, 4)
val mappedStream = inputDataStream.map(new MultiplyByTwoMap())
// 这里创建了一个新的DataStream，其中每个元素都是原DataStream中对应元素的两倍
FlatMapFunction

作用：对每个数据流元素应用一个转换逻辑，可以生成零个、一个或多个输出元素。
import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.api.java.tuple.Tuple2

class TokenizeWords extends FlatMapFunction[String, Tuple2[String, Integer]] {
override def flatMap(value: String, out: Collector[Tuple2[String, Integer]]): Unit = {
for (word <- value.split("\\s+")) {
out.collect(Tuple2(word, 1))
}
}
}

val env = StreamExecutionEnvironment.getExecutionEnvironment
val textStream = env.socketTextStream("localhost", 9999)
val wordCountStream = textStream.flatMap(new TokenizeWords()).keyBy(_._1).sum(1)
// 此处将文本行分割成单词，并生成包含单词与计数值（初始化为1）的元组流
FilterFunction

作用：根据给定条件过滤数据流中的元素。
import org.apache.flink.api.common.functions.FilterFunction

class EvenNumberFilter extends FilterFunction[Int] {
override def filter(value: Int): Boolean = {
// 保留偶数
value % 2 == 0
}
}

val env = StreamExecutionEnvironment.getExecutionEnvironment
val numbersStream = env.fromElements(1, 2, 3, 4, 5)
val evenNumbersStream = numbersStream.filter(new EvenNumberFilter())
// 此处仅保留了数据流中的偶数元素
ReduceFunction

作用：对数据流中的元素进行聚合，按照指定逻辑合并元素。
import org.apache.flink.api.common.functions.ReduceFunction

class SumReducer extends ReduceFunction[Int] {
override def reduce(value1: Int, value2: Int): Int = {
// 对整数进行求和
value1 + value2
}
}

val env = StreamExecutionEnvironment.getExecutionEnvironment
val numbersStream = env.fromElements(1, 2, 3, 4, 5)
val sumResult = numbersStream.reduce(new SumReducer())
// 此处计算数据流中所有整数的总和
此外，还有许多其他高级函数，例如：

KeyedProcessFunction：用于对键控流上的每个键进行状态管理和时间驱动的操作。它可以访问和更新键控状态，并响应定时器事件。这对于实现复杂的事件时间或处理时间相关的业务逻辑非常有用，比如会话窗口处理、状态过期清理等。
以下是一个使用 Scala 实现的 KeyedProcessFunction 示例，该示例展示了如何跟踪每组用户的最后活跃时间，并在用户超过5分钟未活动后发送通知：

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala.OutputTag
import org.apache.flink.util.Collector

// 定义侧输出流标签
val inactiveUsersOutputTag = new OutputTag[String]("inactive-users") {
override def toString: String = "inactive-users"
}

class UserActivityTracker(
timeout: Time
) extends KeyedProcessFunction[String, UserEvent, (String, Boolean)] {

// 定义状态存储用户上次活跃时间
val lastActiveTimeState: ValueState[java.lang.Long] =
getRuntimeContext.getState(new ValueStateDescriptor("last-active-time", classOf[java.lang.Long]))

override def processElement(event: UserEvent, ctx: KeyedProcessFunction[String, UserEvent, (String, Boolean)]#Context, out: Collector[(String, Boolean)]): Unit = {
// 更新用户活跃时间
lastActiveTimeState.update(event.timestamp)

// 注册定时器，在超时后触发
ctx.timerService().registerEventTimeTimer(event.timestamp + timeout.toMilliseconds)
}

override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, UserEvent, (String, Boolean)]#OnTimerContext, out: Collector[(String, Boolean)], out2: Collector[(String, Boolean)] @UnusedParam): Unit = {
// 如果定时器触发，检查是否已超时
if (timestamp > lastActiveTimeState.value()) {
// 用户已超时未活跃
val userId = ctx.getCurrentKey
// 输出到侧输出流
out2.collect(userId)

// 可选地，也可以清除或更新状态
lastActiveTimeState.clear()
}
}
}

case class UserEvent(userId: String, timestamp: Long)

// 创建环境和数据源...
val env = StreamExecutionEnvironment.getExecutionEnvironment
val eventsStream = ...

// 应用 KeyedProcessFunction
val userActivities = eventsStream
.keyBy(_.userId)
.process(new UserActivityTracker(Time.minutes(5)))

// 获取侧输出流
val inactiveUsers = userActivities.getSideOutput(inactiveUsersOutputTag)

// 分别处理主输出流（活跃用户相关处理）和侧输出流（不活跃用户的通知）
inactiveUsers.print() // 假设打印不活跃用户ID

WindowFunction：一种专门用来处理窗口数据的函数，它可以在窗口触发时对窗口内的所有元素进行一次性的处理。窗口可以基于时间（如每5分钟一个窗口）或数量（如每100个元素一个窗口）进行划分。
以下是一个使用 Scala 实现的 WindowFunction 示例，我们将计算每5分钟一组用户的点击次数：

import org.apache.flink.api.common.functions.WindowFunction
import org.apache.flink.api.java.tuple.Tuple2
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

case class ClickEvent(userId: String, timestamp: Long)

class ClickCounter extends WindowFunction[ClickEvent, Tuple2[String, Int], String, TimeWindow] {

override def apply(key: String, window: TimeWindow, inputs: Iterable[ClickEvent], out: Collector[Tuple2[String, Int]]): Unit = {
// 计算特定用户在给定时间窗口内的点击次数
val clickCount = inputs.size

// 输出用户ID和点击次数
out.collect(Tuple2(key, clickCount))
}
}

// 创建环境和数据源...

本文系转载，版权归原作者所有，如若侵权请联系我们进行删除！  

《行业指标体系白皮书》下载地址：https://www.dtstack.com/resources/1057/?src=bbs

《数据治理行业实践白皮书》下载地址：https://www.dtstack.com/resources/1001/?src=bbs

《数栈V6.0产品白皮书》下载地址：https://www.dtstack.com/resources/1004/?src=bbs

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs

同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack