本章主要介绍Flink的类型系统和支持的数据类型并介绍数据转换（data transformation）和分区转换（partition transformation）。
构建一个Flink流式数据需要以下几步：
- 设置执行环境
- 从数据源中读取一条或者多条流
- 通过一系列流式转换来实现应用逻辑
- 选择性地将结果输出到一个或者多个数据汇中。
- 执行程序
设置执行环境
DataStream API执行的环境如果使用
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
在调用时，会根据上下文判断是否返回一个本地或者远程环境。当然，也可以创建本地执行环境：
val localEnv: LocalStreamEnvironment = StreamExecutionEnvironment.createLocalEnvironment()
也可以创建远程的执行环境：
StreamExecutionEnvironment.createRemoteEnvironment(
"host", //JobManager的主机名
12456, //JobManager的端口号
"path/jarFile.jar" //需要传输的jar包的路径
)
读取输入流
应用转换
输出结果
转化操作
完成一个DataStream API程序在本质上可以归纳为：通过组合不同的转换来创建一个满足应用逻辑的Dataflow图.
将DataStream API的转换分为4类：
- 作用域单个事件的基本转换
- 针对相同键值事件的KeyedStream转换。
- 将多条数据流合并为一条或者一条数据流拆分为多条流的转换。
- 对流中的事件进行重新组织的分发转换。
基本转换
- Map转换
在map转换算子中，最重要的就是MapFunction函数了。下面就是一个简单的映射器，他会提取输入流中的每一个SensorReading记录的第一个字段di
package com.fengjinxin
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.environment._
import org.apache.flink.streaming.api.scala.{DataStream, createTypeInformation}
object flinkTest02 {
def main(args: Array[String]): Unit = {
//TODO 开启环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//TODO 数据操作
//分配时间戳和水位线
val readings: DataStream[SensorReading] = env.addSource(new SensorSource).assignTimestampsAndWatermarks(new SensorTimeAssigner)
val sensorIds: DataStream[String] = readings.map(new MyMapFunction)
sensorIds.print()
//TODO 执行程序
env.execute("map function")
}
}
//继承MapFunction
class MyMapFunction extends MapFunction[SensorReading, String] {
override def map(t: SensorReading): String = t.id
}
case class SensorReading(id:String, timestamp:Long, temperature:Double)
- FlatMap
flatMap类似于map，但可以对每个输入事件产生0个或者多个输出事件。对应的FlatMapFunction定义了flatMap方法，可以在其中通过向Collector对象传递数据的方式返回多个时间作为结果。
下面将一句话分割成多个单词：
object flinkTest03 {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//TODO 数据操作
val sentences:DataStream[String]=env.addSource(new SourceDataStream)
val words: DataStream[String] = sentences.flatMap(new myFlatMapFunction)
words.print()
//TODO 执行程序
env.execute("flatMap function")
}
}
class myFlatMapFunction extends FlatMapFunction[String, String] {
override def flatMap(t: String, collector: Collector[String]): Unit = t.split(" ")
}
- 基于KeyedStream的转换
在很多业务中，很多事件需要按照某个属性分组后进行处理，则需要DataStream->KeyedStream->DataStream进行操作。
下面将DataSream进行keyBy转化将一个DataStream转化为KeyedStream，然后对他进行滚动聚合操作以及reduce操作
object flinkTest04 {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//设置环境为事件时间
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//TODO 数据操作
val readings:DataStream[SensorReading]=env.addSource(new SensorSource)
.assignTimestampsAndWatermarks(new SensorTimeAssigner)
val keyed: KeyedStream[SensorReading, String] = readings.keyBy(r => r.id)
keyed.print()
//TODO 程序执行
env.execute("Keyed Stream")
}
}
case class SensorReading(id:String, timestamp:Long, temperature:Double)
- 滚动聚合
滚动聚合转换作用于KeyedStream上，将生成一个包含聚合结果的DataStream。算是这样一个过程：KeyedStream->DataStream。
滚动聚合不需要用户去自定义函数，但是需要接收一个用于指定聚合目标字段的参数。DataStream API中提供了一下滚动聚合的方法：
- sum():注意对于POJO对象，sum等转换操作中的参数必须与POJO的字段保持一致。
val keyed: KeyedStream[SensorReading, String] = readings.keyBy(r => r.id)
val sumStream: DataStream[SensorReading] = keyed.
timeWindow(Time.seconds(5)).sum("timestamp")
sumStream.print()
- min()
val keyed: KeyedStream[SensorReading, String] = readings.keyBy(r => r.id)
val sumStream: DataStream[SensorReading] = keyed.
timeWindow(Time.seconds(5)).min("timestamp")
sumStream.print()
- max()
- minBy()
- maxBy()
- Reduce
reduce转化是滚动聚合转化的泛化。将一个ReduceFunction应用在一个KeyedStream上，每个到来的事件都会和reduce结果进行一次组合，从而产生一个新的DataStream。reduce转换不会改变数据类型。
//ReduceFunction的元素类型
ReduceFunction[T]
> reduce(T, T): T
注意：只对有限键值域使用滚动reduce操作:滚动 reduce算子会为每个处理过的键值维持一个状态。由于这些状态不会被自动清理,所以该算子只能用于键值域有限的流。
多流转换
- Union转换
Union的部分源码如下：
public final DataStream<T> union(DataStream<T>... streams) {
List<Transformation<T>> unionedTransforms = new ArrayList();
unionedTransforms.add(this.transformation);
DataStream[] var3 = streams;
int var4 = streams.length;
for(int var5 = 0; var5 < var4; ++var5) {
DataStream<T> newStream = var3[var5];
if (!this.getType().equals(newStream.getType())) {
throw new IllegalArgumentException("Cannot union streams of different types: " + this.getType() + " and " + newStream.getType());
}
unionedTransforms.add(newStream.getTransformation());
}
return new DataStream(this.environment, new UnionTransformation(unionedTransforms));
}
也就是说Union连接的是相同类型的DataStream。
object flinkTest05 {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//TODO 数据操作
//数据
val sensorReading1: DataStreamSource[SensorReading] = env.fromElements(SensorReading("1", 12L, 12.5), SensorReading("2", 5L, 51.5),
SensorReading("3", 15L, 14.6), SensorReading("4", 45L, 18.2)
)
val sensorReading2:DataStreamSource[SensorReading]=env.fromElements(SensorReading("5", 12L, 12.5),SensorReading("6", 5L, 51.5),
SensorReading("7", 15L, 14.6), SensorReading("8", 45L, 18.2)
)
//转化
val sensorReading3: DataStream[SensorReading] = sensorReading1.union(sensorReading2)
//输出
sensorReading3.print()
//TODO 执行程序
env.execute()
}
}
case class SensorReading(id:String, timestamp:Long, temperature:Double)
- Connect, coMap, coFlatMap
上面的Union中连接的是相同数据类型的DataStream。在业务中尝尝遇到不同数据类型的DataStream进行连接，这时候connect转换就起作用了。
object Main {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//TODO 数据操作
//数据
val intStream: DataStream[Int] = env.fromElements(1, 2, 3, 4, 5)
val strStream: DataStream[String] = env.fromElements("kone", "bob", "alices")
//转换
val connectStream: ConnectedStreams[Int, String] = intStream.connect(strStream)
//输出
//TODO 执行程序
env.execute()
}
}
Connectedstreams对象提供了map()和f1 atap()方法,它们分别接收一个Comapfunction和一个 Coflatmap Function作为参数。
注意：CoMapFunction和 CoFlatmapFunction内方法的调用顺序无法控制。一旦对应流中有事件到来,系统就需要调用相应的方法。
上面的connect只是简单的将两个DataStream进行连接。在很多业务中，我们希望两个DataStream在连接时按照键值分区或者连接两个已经按照键值分好区的数据流。
object Main {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//TODO 数据操作
//数据
val one: DataStreamSource[(Int, Long)] = env.fromElements((1, 2L), (2, 2L))
val two: DataStreamSource[(Int, String)] = env.fromElements((1, "kone"), (2, "bob"))
//转换
// 对两个联结后的数据流按键值分区
val keyedConnect1: ConnectedStreams[(Int, Long), (Int, String)] = one.connect(two).keyBy(0, 0)
//联结两个已经按照键值分好区的数据流
val keyedConnect2: ConnectedStreams[(Int, Long), (Int, String)] = one.keyBy(0).connect(two.keyBy(0))
//输出
//TODO 程序执行
env.execute()
}
}
在业务中还会遇到两个DataStream的数量级相差较大。这个时候为了提高程序的效率，常常将数量级小的DataStream设置为广播。调用：broadcast
object Main {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//TODO 数据操作
//数据
val one: DataStreamSource[(Int, Long)] = env.fromElements((1, 2L), (2, 2L))
val two: DataStreamSource[(Int, String)] = env.fromElements((1, "kone"), (2, "bob"))
//转换
val broadcastConnect: ConnectedStreams[(Int, Long), (Int, String)] = one.connect(two.broadcast())
//输出
//TODO 程序执行
env.execute()
}
}
所有广播流的事件都会被复制多份并分别发往后续处理函数所在算子的每个实例;而所有非广播流的事件只是会被简单地转发。这样一来,我们就可以联合处理两个输入流的元素。
Split和Select
上文的Union和Connect都是将多个DataStream转化成一个DataStream。而Split就是将一个DataStream分成0个，1个或者多个DataStream。DataStream.split()方法接收一个OutputSelector,它用来定义如何将数据流的元素分别到不同的命名输出中。
Outputselector中定义的 select()方法会在毎个输入事件到来时被调用,并随即返回一个java.lang. Iterable[String]对象。针对某记录所返回的一系列 String值指定了该记录需要被发往哪些输出流。
object Main {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
//TODO 数据操作
//数据
val inputStream: DataStream[(Int, String)] = env.fromElements((1, "kone"), (2, "alices"), (3, "kiwen"), (4, "bob"), (5, "hik"))
//转换
val splitted: SplitStream[(Int, String)] = inputStream.split(t => if (t._1 > 2) Seq("low") else Seq("high"))
// 开始分开DataStream
val low: DataStream[(Int, String)] = splitted.select("low")
val high: DataStream[(Int, String)] = splitted.select("high")
val all: DataStream[(Int, String)] = splitted.select("low", "high")
//输出
low.print("low:")
high.print("high:")
all.print("all:")
//TODO 执行程序
env.execute("split app")
}
}
分发转换
在构建DataStream API程序时，系统会根据语义和配置自动的选择数据分区的策略并将数据转发到正确的目标。有些时候，我们希望能够在应用级别控制这些分区策略，或者自定义分区器。
注意, keyBy()和本节介绍的分发转换不同。所有本节介绍的转换都会生成一个 Datastream,而 key By()会生成一个 Keyedstream。基于后者可以应用那些能够访问键值分区状态的转换。
- 随机：利用DataStream.shuffle()方法实现随机数据交换策略。
- 轮流：rebalance()方法会将输入流中的事件以轮流方式均匀分配给后继任务。
- 重调：rescale()也会以轮流方式对事件进行分发。
rebalance()和 rescale()的本质不同体现在生成任务连接的方式。rebalance()会在所有发送任务和接收任务之间建立通信通道;而rescale()中每个发送任务只会和下游算子的部分任务建立通道。下图展示了两者之间的区别：
- 全局：global()方法会将输入流中的所有事件发往下游算子的第一个并行任务。
- 自定义：利用partitionCustom()方法来定义分区策略。该方法接收一个Partitioner对象，用户可以实现分区逻辑，定义分区需要参照的字段或键值位置。
设置并行度
在一般情况下，最好将算子并行度设置为随环境默认并行度变化的值。这样就可以通过提交客户端来轻易调度并行度，从而实现应用的扩缩容。
- 获取默认的并行度：
object Main {
def main(args: Array[String]): Unit = {
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
val defultP: Int = env.getParallelism
}
}
- 同时也可以在代码中设置并行度，并且一旦设置将无法通过客户端控制应用并行度：(注意这里设置的是整个环境的并发度)
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(32)
- 指定某个算子的并发度：比如下面只是指定map的并发度：
object Main {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
val defaultP: Int = env.getParallelism
//TODO 数据操作
//数据
val dataStream: DataStream[Student] = env.fromCollection(List(Student(1, "kone"), Student(2, "bob")))
//转换
val addIdStudentStream: DataStream[Student] = dataStream.map(t => Student(t.id + 1, t.name)).setParallelism(defaultP * 2) //将map的并行度设置为2
//输出
addIdStudentStream.print("id + 1：")
//TODO 程序执行
env.execute()
}
}
case class Student(id:Int, name:String)
类型
在flink中需要对数据进行网络传输，所以需要进行序列化与反序列化的过程。Flink利用类型信息的概念来表示数据类型，并且对于每种类型，都会为其生成特定的序列化器，反序列化器以及比较器。
此外, Flink中还有一个类型提取系统,它可以通过分析函数的输入、输出类型来自动获取类型信息,继而得到相应的序列化器和反序列化器。但在某些情况下,例如使用了 Lambda函数或泛型类型,则必须显式指定类型信息才能启动应用或提高其性能。
支持的数据类型
Flink支持的类型有以下几种：
注意,如果可能请尽量避免使用Kryo。作为一个通用序列化器,Kryo的效率通常不高。为了提高效率,Fink提供配置选项可以提前将类在Kryo中注册好。此外,对于数据类型发生改变的情况,Kryo没有提供很好的迁移方案。
下面详细分析各个类型
- 原始类型：
- Java和Scala元组
Flink提供了Java元组的高效实现,它最多可包含25个字段,每个字段长度都对应一个单独的实现类一 Tuple1、 Tuple2,直到 Tuple25。这些元组类都是强类型的。
- Scala样例类
- POJO类型
Flink会分析哪些不属于任何一类的数据类型，并尝试将他们作为POJO类型进行处理。如果满足以下条件，Flink就会将他看做POJO类型：
- 是一个公有类
- 有一个公有的参数默认构造函数。
- 所有字段都是公有的或者提供对应的get和set函数。并且这些函数遵循默认的命名规范。比如对于X命名为getX和setX。
- 所有字段类型都是Flink所支持的。
- 数组，列表，映射，枚举以及其他特殊类型：
Flink支持多种具有特殊用途的类型,例如:原始或对象类型的数组, Java的 Arraylist、 Hashmap及Enum, Hadoop的 Writable类型等。此外, 它还为 Scala I的 Either、 option、Try类型以及 Flink内部Java版本的Either类型提供了相应的类型信息。
为数据累次那个创建类型信息
Flink类型系统的核心类是TypeInfromation，它为系统生成序列化器和比较器提供了必要的信息。当应用执行时，Flink的类型系统会为将来所需处理的每种类型自动推断TypeInformation。一个名为类型提取器的组件会分析所有函数的泛化类型及返回类型，以获得对应的TypeInformation对象。
- 在Java中通过org.apache.flink.api.common.typeInfo.Types
// Java元组的TypeInformation
TypeInformation<Tuple2<long, String> >tupleType=Types.TUPLE(Types.LONG, Types.STRING);
// POJO的TypeInformation
TypeInformation<Person> personType=Types.POJO(Person.class);
- 在Scala中通过org.apache.flink.api.scala.typeutils.Types
object Main {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
val defaultP: Int = env.getParallelism
//TODO 数据操作
//原始类型的TypeInformation
val stringType: TypeInformation[String] = Types.STRING
//Scala元组的YupeInformation
val tupleType: TypeInformation[(Int, String)] = Types.TUPLE[(Int, String)]
//样例类的TypeInformation
val caseClassType: TypeInformation[Student] = Types.CASE_CLASS[Student]
//TODO 程序执行
env.execute()
}
}
case class Student(id:Int, name:String)
显式提供类型信息
在很多情况下，TypeInformation能够正确推断出类型。但是有时候有些必要的信息无法提取到或者使用了TypeInfromation无法生成最高效的序列化器和反序列器。因此，这时候就需要向Flink显式提供TypeInformation对象。
提供TypeInformation的方法有以下两种：
- 通过ResultTypeQueryable接口来扩展函数，在其中提供返回类型的TypeInformation。
case class Student(id:Int, name:String)
class Tuple2ToStudent extends MapFunction[(Int, String), Student]
with ResultTypeQueryable[Student]{
override def map(t: (Int, String)): Student = Student(t._1, t._2)
//为输出数据类型提供TypeInformation
override def getProducedType: TypeInformation[Student] = Types.CASE_CLASS[Student]
}
- 还可以在定义Dataflow的时候使用Java DataStream API中的returns方法
DataStream<Tuple2<String, Integer> > tuples= ...
DataStream<Student> students=tuples.map(t-> new Student(t.f0, t.f1))
//为map lambda函数的返回类型提供TypeInformation
.returns(Types.POJO(Person.class));
定义键值和引用手段
Flink中将键值定义为输入数据上的函数。
下面将讨论基于数据类型定义引用字段和键值的几种方法。
- 字段位置：使用元祖对应元素的字段位置来定义键值。
object Main {
def main(args: Array[String]): Unit = {
//TODO 创建环境
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//TODO 数据操作
//数据
val dataList = List((1, 1L, "kone"), (2, 1L, "bob"), (3, 12L, "alices"))
val dataStream: DataStream[(Int, Long, String)] = env.fromCollection(dataList)
//转换
//下面将DataStream中元组的第2个字段位置作为输入流
val keyed2: KeyedStream[(Int, Long, String), Tuple] = dataStream.keyBy(1)
//下面将DataStream中元组的第2个和第3个位置作为输入流
val keyed23: KeyedStream[(Int, Long, String), Tuple] = dataStream.keyBy(1, 2)
//输出
keyed2.print("将位置2的作为输入流：")
keyed23.print("将位置2，3作为输入流：")
//TODO 程序执行
env.execute("Keyed Test")
}
}
- 字段表达式：可用于元组，POJO和样例类。
studentStream.keyBy("id")
//对于元组
dataStream.keyBy("2") //以第3个位置
dataStream.keyBy("_1") //以第1个位置
- 键值选择器：使用KeySelector函数。
val dataList = List((1, 1L, "kone"), (2, 1L, "bob"), (3, 12L, "alices"))
val dataStream: DataStream[(Int, Long, String)] = env.fromCollection(dataList)
//转换
val key1: KeyedStream[(Int, Long, String), Int] = dataStream.keyBy(r => r._1) //第一个位置
实现函数
Flink找那个所有用户自定义函数（如MapFuncton，FilterFunction）的接口都是以接口或抽象类的形式对外暴露。
lambda函数
//数据
val dataStream: DataStream[String] = env.fromElements("kone", "bob")
//转化
val filterData: DataStream[String] = dataStream.filter(_.contains("kone"))
//输出
filterData.print("输出：")
富函数
DataStream API中所有的转换函数都有对应的富函数。富函数的使用位置和普通函数以及Lambda函数相同。富函数的命名规则以Rich开头，后面跟着普通转换函数的名字。
导入外部和Flink依赖
大多数Flink应用需要需要其他库比如Kafka的连接器库。