Flink编程模型

数据集类型

1. 有界数据集 有界数据集有时间边界。在处理过程中，数据将在一定时间内开始和结束，可能是一分钟或一天内的交易数据。对有界数据集的数据处理称为批计算（Batch Processing），例如，从数据RDBMS或在文件系统等系统中读取，然后在分布式系统中处理，最后将处理结果写入存储介质，整个过程称为批处理过程。对于批数据处理，业内流行的分布式批处理框架有Apache Hadoop和Apache Spark等。
2. 无界数据集 对于无界数据集，数据从生成开始就不断生成新数据，因此数据没有边界，如服务器日志、传感器信号数据等。对应于批量数据处理，无界数据集的数据处理称为流式数据处理，称为流处理（Streaming Process）。可以看出，流量数据处理过程将更加复杂，因为需要考虑处理过程中的数据顺序混乱和系统容错，因此流量处理需要使用特殊的流量数据处理技术。目前业界的Apache Storm、Spark Streaming、Apache Flink等分布式计算引擎可以不同程度地支持流式数据的处理。
3. 统一数据处理 有界数据集和无界数据集只是一个相对的概念，主要取决于时间范围。可以认为，一段时间内的无界数据集实际上是有界数据集，有界数据也可以通过某些方法转换为无界数据。例如，系统一年的订单交易数据本质上应该是有界数据集，但当我们将其逐一发送到流式系统，并通过流式系统处理数据时，我们可以认为数据相对无界。无界数据也可以分为有界数据进行处理。例如，将系统生成的数据连接到存储系统，按年或月切割，切割成不同时间长度的有界数据集，然后通过批处理处理数据。从以上分析可以得出结论：有界数据和无界数据实际上可以相互转换。有了这样的理论基础，行业还提出了不同的计算框架，可以统一数据处理不同的数据类型。 目前，在业内知名的开源大数据处理框架中，流式计算和批量计算可以同时支持。典型的代表是Apache Spark和Apache Flink两套框架。其中Spark不同类型的数据集通过批处理模式统一处理，流数据按批次分为微批（有界数据集）。Flink通过流处理模式，从另一个角度统一处理不同类型的数据集。Flink流式数据以更符合数据生成规律的方式处理，有界数据可以转换为无界数据统一流式，最终在一套流式引擎中统一批处理和流处理，使用户可以使用一套引擎进行批计算和流计算任务。 前面提到过，用户可能需要并行使用各种计算框架来解决不同类型的数据处理，例如，用户可以使用它Flink使用流计算引擎作为流计算引擎Spark或者MapReduce作为仅增加了系统的复杂性，也增加了用户学习和运维的成本。而Flink作为一个新兴的分布式计算引擎，它可以在统一的平台上处理流动任务和批量任务。同时，流动计算模式的使用更符合数据生成的规律Flink未来将成为众多大数据处理引擎的明星。

Flink编程接口

Flink程序结构

package com.realtime.flink.streaming import org.apache.flink.api.java.utils.ParameterTool import org.apache.flink.streaming.api.scala.{DataStream, StreamExecution Environment, _} object WordCount {   def main(args: Array[String]) {     // 第一步:设置执行环境     val env = StreamExecutionEnvironment.getExecutionEnvironment     // 第二步：指定数据源地址，读取输入数据     val text = env.readTextFile("file:///path/file")     // 第三步:指定数据集转换操作逻辑     val counts: DataStream[(String, Int)] = text         .flatMap(_.toLowerCase.split(" "))       .filter(_.nonEmpty)       .map((_, 1))       .keyBy(0)       .sum(1)     // 第四步：指定计算结果的输出位置     if (params.has("output")) {       counts.writeAsText(params.get("output"))     } else {       println("Printing result to stdout. Use --output to specify output path.")       counts.print()     }     // 第五步：指定名称并触发流式任务     env.execute("Streaming WordCount")   } } 

//设定Flink运行环境，如果在本地启动则创建本地环境，如果是在集群上启动，则创建集群环境
StreamExecutionEnvironment.getExecutionEnvironment
//指定并行度创建本地执行环境
StreamExecutionEnvironment.createLocalEnvironment(5)
//指定远程JobManagerIP和RPC端口以及运行程序所在jar包及其依赖包
StreamExecutionEnvironment.createRemoteEnvironment("JobManagerHost",6021,5,"/user/application.jar")

//设定Flink运行环境，如果在本地启动则创建本地环境，如果是在集群上启动，则创建集群环境
ExecutionEnvironment.getExecutionEnvironment
//指定并行度创建本地执行环境
ExecutionEnvironment.createLocalEnvironment(5)
//指定远程JobManagerIP和RPC端口以及运行程序所在jar包及其依赖包
ExecutionEnvironment.createRemoteEnvironment("JobManagerHost",6021,5,"/user/application.jar")

val text:DataStream[String] = env.readTextFile("file:///path/file")

通过读取文件并转换为DataStream[String]数据集，这样就完成了从本地文件到分布式数据集的转换，同时在Flink中提供了多种从外部读取数据的连接器，包括批量和实时的数据连接器，能够将Flink系统和其他第三方系统连接，直接获取外部数据。

3. 执行转换操作 数据从外部系统读取并转换成DataStream或者DataSet数据集后，下一步就将对数据集进行各种转换操作。Flink中的Transformation操作都是通过不同的Operator来实现，每个Operator内部通过实现Function接口完成数据处理逻辑的定义。在DataStream API和DataSet API提供了大量的转换算子，例如map、flatMap、filter、keyBy等，用户只需要定义每种算子执行的函数逻辑，然后应用在数据转换操作Dperator接口中即可。如下代码实现了对输入的文本数据集通过FlatMap算子转换成数组，然后过滤非空字段，将每个单词进行统计，得到最后的词频统计结果。

val counts: DataStream[(String, Int)] = text
    .flatMap(_.toLowerCase.split(" "))//执行FlatMap转换操作
  .filter(_.nonEmpty)//执行Filter操作过滤空字段
  .map((_, 1))//执行map转换操作，转换成key-value接口
  .keyBy(0)//按照指定key对数据重分区
  .sum(1)//执行求和运算操作

在上述代码中，通过Scala接口处理数据，极大地简化数据处理逻辑的定义，只需要通过传入相应Lambada计算表达式，就能完成Function定义。特殊情况下用户也可以通过实现Function接口来完成定义数据处理逻辑。然后将定义好的Function应用在对应的算子中即可。Flink中定义Funciton的计算逻辑可以通过如下几种方式完成定义。 （1）通过创建Class实现Funciton接口 Flink中提供了大量的函数供用户使用，例如以下代码通过定义MyMapFunction Class实现MapFunction接口，然后调用DataStream的map()方法将MyMapFunction实现类传入，完成对实现将数据集中字符串记录转换成大写的数据处理。

val dataStream: DataStream[String] = env.fromElements("hello", "flink")
dataStream.map(new MyMapFunction)
class MyMapFunction extends MapFunction[String, String] {
  override def map(t: String): String = {
    t.toUpperCase()
  }
}

（2）通过创建匿名类实现Funciton接口 除了以上单独定义Class来实现Function接口之处，也可以直接在map()方法中创建匿名实现类的方式定义函数计算逻辑。

val dataStream: DataStream[String] = env.fromElements("hello", "flink")
//通过创建MapFunction匿名实现类来定义Map函数计算逻辑
  dataStream.map(new MapFunction[String, String] {
    //实现对输入字符串大写转换
    override def map(t: String): String = {
      t.toUpperCase()
    }
  })

（3）通过实现RichFunciton接口 前面提到的转换操作都实现了Function接口，例如MapFunction和FlatMap-Function接口，在Flink中同时提供了RichFunction接口，主要用于比较高级的数据处理场景，RichFunction接口中有open、close、getRuntimeContext和setRuntimeContext等方法来获取状态，缓存等系统内部数据。和MapFunction相似，RichFunction子类中也有RichMapFunction，如下代码通过实现RichMapFunction定义数据处理逻辑，具体的RichFunction的介绍可以参考后续文章。

//定义匿名类实现RichMapFunction接口，完成对字符串到整形数字的转换
data.map (new RichMapFunction[String, Int] {
  def map(in: String):Int = { in.toInt }
})

4. 分区Key指定 在DataStream数据经过不同的算子转换过程中，某些算子需要根据指定的key进行转换，常见的有join、coGroup、groupBy类算子，需要先将DataStream或DataSet数据集转换成对应的KeyedStream和GroupedDataSet，主要目的是将相同key值的数据路由到相同的Pipeline中，然后进行下一步的计算操作。需要注意的是，在Flink中这种操作并不是真正意义上将数据集转换成Key-Value结构，而是一种虚拟的key，目的仅仅是帮助后面的基于Key的算子使用，分区Key可以通过两种方式指定：

（1）根据字段位置指定

在DataStream API中通过keyBy()方法将DataStream数据集根据指定的key转换成重新分区的KeyedStream，如以下代码所示，对数据集按照相同key进行sum()聚合操作。

val dataStream: DataStream[(String, Int)] = env.fromElements(("a", 1), ("c",
  2))
//根据第一个字段重新分区，然后对第二个字段进行求和运算
Val result = dataStream.keyBy(0).sum(1)

在DataSet API中，如果对数据根据某一条件聚合数据，对数据进行聚合时候，也需要对数据进行重新分区。如以下代码所示，使用DataSet API对数据集根据第一个字段作为GroupBy的key，然后对第二个字段进行求和运算。

val dataSet = env.fromElements(("hello", 1), ("flink", 3))
  //根据第一个字段进行数据重分区
  val groupedDataSet:GroupedDataSet[(String,Int)] = dataSet.groupBy(0)
  //求取相同key值下第二个字段的最大值
  groupedDataSet.max(1)

（2）根据字段名称指定 KeyBy和GroupBy的Key除了能够通过字段位置来指定之外，也可以根据字段的名称来指定。使用字段名称需要DataStream中的数据结构类型必须是Tuple类或者POJOs类的。如以下代码所示，通过指定name字段名称来确定groupby的key字段。

val personDataSet = env.fromElements(new Persion("Alex", 18),new 
Persion("Peter", 43))
//指定name字段名称来确定groupby字段
personDataSet.groupBy("name").max(1)

如果程序中使用Tuple数据类型，通常情况下字段名称从1开始计算，字段位置索引从0开始计算，以下代码中两种方式是等价的。

val personDataStream = env.fromElements(("Alex", 18),("Peter", 43))
//通过名称指定第一个字段名称
personDataStream.keyBy("_1")
//通过位置指定第一个字段
personDataStream.keyBy(0)

如果在Flink中使用嵌套的复杂数据结构，可以通过字段名称指定Key，例如：

class CompelexClass(var nested: NestedClass, var tag: String) {
  def this() { this(null, "") }
}
class NestedClass (
    var id: Int,
    tuple: (Long, Long, String)){
  def this() { this(0, (0, 0, "")) }
}

通过调用“nested”获取整个NestedClass对象里所有的字段，调用“tag”获取CompelexClass中tag字段，调用“nested.id”获取NestedClass中的id字段，调用“nested.tuple._1”获取NestedClass中tuple元祖的第一个字段。由此可以看出，Flink能够支持在复杂数据结构中灵活地获取字段信息，这也是非 Key-Value的数据结构所具有的优势。

（3）通过Key选择器指定 另外一种方式是通过定义Key Selector来选择数据集中的Key，如下代码所示，定义KeySelector，然后复写getKey方法，从Person对象中获取name为指定的Key。

case class Person(name: String, age: Int)
val person= env.fromElements(Person("hello",1), Person("flink",4))
//定义KeySelector,实现getKey方法从case class中获取Key
val keyed: KeyedStream[WC]= person.keyBy(new KeySelector[Person, String]() {
  override def getKey(person: Person): String = person.word
})

5. 输出结果 数据集经过转换操作之后，形成最终的结果数据集，一般需要将数据集输出在外部系统中或者输出在控制台之上。在Flink DataStream和DataSet接口中定义了基本的数据输出方法，例如基于文件输出writeAsText()，基于控制台输出print()等。同时Flink在系统中定义了大量的Connector，方便用户和外部系统交互，用户可以直接通过调用addSink()添加输出系统定义的DataSink类算子，这样就能将数据输出到外部系统。以下实例调用DataStream API中的writeAsText()和print()方法将数据集输出在文件和客户端中。

//将数据输出到文件中
counts.writeAsText("file://path/to/savefile")
//将数据输出控制台
counts.print()

6. 程序触发 所有的计算逻辑全部操作定义好之后，需要调用ExecutionEnvironment的execute()方法来触发应用程序的执行，其中execute()方法返回的结果类型为JobExecutionResult，里面包含了程序执行的时间和累加器等指标。需要注意的是，execute方法调用会因为应用的类型有所不同，DataStream流式应用需要显性地指定execute()方法运行程序，如果不调用则Flink流式程序不会执行，但对于DataSet API输出算子中已经包含对execute()方法的调用，则不需要显性调用execute()方法，否则会出现程序异常。

//调用StreamExecutionEnvironment的execute方法执行流式应用程序
env.execute("App Name");

Flink数据类型

数据类型支持

//创建Int类型的数据集
val intStream:DataStream[Int] = env.fromElements(3, 1, 2, 1, 5)
//创建String类型的数据集
val dataStream: DataStream[String] = env.fromElements("hello", "flink")

//通过从数组中创建数据集
val dataStream: DataStream[Int] = env.fromCollection(Array(3, 1, 2, 1, 5))
//通过List集合创建数据集
val dataStream: DataStream[Int] = env.fromCollection(List(3, 1, 2, 1, 5))

//通过实例化Tuple2创建具有两个元素的数据集
val tupleStream2: DataStream[Tuple2[String, Int]] = env.fromElements(new Tuple2("a",1), new Tuple2("c", 2))

//定义WordCount Case Class数据结构
case class WordCount(word: String, count: Int)
//通过fromElements方法创建数据集
val input = env.fromElements(WordCount("hello", 1), WordCount("world", 2))
val keyStream1 = input.keyBy("word") // 根据word字段为分区字段，
val keyStream2 = input.keyBy(0) //也可以通过指定position分区

//通过scala Tuple创建具有两个元素的数据集
val tupleStream: DataStream[Tuple2[String, Int]] = env.fromElements(("a", 1),
  ("c", 2))
//使用默认字段名称获取字段，其中_1表示tuple这种第一个字段
tupleStream.keyBy("_1")

//定义Java Person类，具有public修饰符
public class Person {
  //字段具有public修饰符
    public String name;
    public int age;
  //具有默认空构造器
    public Person() {
    }
    public Person(String name, int age) {
      this.name = name;
      this.age = age;   
    }
}

val persionStream = env.fromElements(new Person("Peter",14),new 
Person("Linda",25))
//通过Person.name来指定Keyby字段
persionStream.keyBy("name")
Scala POJOs数据结构定义如下，使用方式与Java POJOs相同。
class Person(var name: String, var age: Int) {
    //默认空构造器
      def this() {
        this(null, -1)
  }
}

//创建Map类型数据集
val mapStream = 
env.fromElements(Map("name"->"Peter","age"->18),Map("name"->"Linda",
"age"->25))
//创建List类型数据集
val listStream = env.fromElements(List(1,2,3,5),List(2,4,3,2))

TypeInformation信息获取

1. Scala API类型信息 Scala API通过使用Manifest和类标签，在编译器运行时获取类型信息，即使是在函数定义中使用了泛型，也不会像Java API出现类型擦除的问题，这使得Scala API具有非常精密的类型管理机制。同时在Flink中使用到Scala Macros框架，在编译代码的过程中推断函数输入参数和返回值的类型信息，同时在Flink中注册成TypeInformation以支持上层计算算子使用。 当使用Scala API开发Flink应用，如果使用到Flink已经通过TypeInformation定义的数据类型，TypeInformation类不会自动创建，而是使用隐式参数的方式引入，代码不会直接抛出编码异常，但是当启动Flink应用程序时就会报“could not find implicit value for evidence parameter of type TypeInformation”的错误。这时需要将TypeInformation类隐式参数引入到当前程序环境中，代码实例如下：

import org.apache.flink.api.scala._

2. Java API类型信息 由于Java的泛型会出现类型擦除问题，Flink通过Java反射机制尽可能重构类型信息，例如使用函数签名以及子类的信息等。同时类型推断在当输出类型依赖于输入参数类型时相对比较容易做到，但是如果函数的输出类型不依赖于输入参数的类型信息，这个时候就需要借助于类型提示（Ctype Himts）来告诉系统函数中传入的参数类型信息和输出参数信息。如代码清单3-3通过在returns方法中传入TypeHint实例指定输出参数类型，帮助Flink系统对输出类型进行数据类型参数的推断和收集。

DataStream<Integer> typeStream = input
  .flatMap(new MyMapFunction<String, Integer>())
  .returns(new TypeHint<Integer>() {//通过returns方法指定返回参数类型
    });
//定义泛型函数，输入参数类型为<T,O>，输出参数类型为O
  class MyMapFunction<T, O> implements MapFunction<T, O> {
    public void flatMap(T value, Collector<O> out) {
      //定义计算逻辑
    }
  }

在使用Java API定义POJOs类型数据时，PojoTypeInformation为POJOs类中的所有字段创建序列化器，对于标准的类型，例如Integer、String、Long等类型是通过Flink自带的序列化器进行数据序列化，对于其他类型数据都是直接调用Kryo序列化工具来进行序列化。 通常情况下，如果Kryo序列化工具无法对POJOs类序列化时，可以使用Avro对POJOs类进行序列化，如下代码通过在ExecutionConfig中调用enableForceAvro()来开启Avro序列化。

ExecutionEnvironment env = 
ExecutionEnvironment.getExecutionEnvironment();
//开启Avro序列化方式
env.getConfig().enableForceAvro();

如果用户想使用Kryo序列化工具来序列化POJOs所有字段，则在ExecutionConfig中调用enableForceKryo()来开启Kryo序列化。

final ExecutionEnvironment env = 
ExecutionEnvironment.getExecutionEnvironment();
env.getConfig().enableForceKryo();

如果默认的Kryo序列化类不能序列化POJOs对象，通过调用ExecutionConfig的addDefault-KryoSerializer()方法向Kryo中添加自定义的序列化器。

env.getConfig().addDefaultKryoSerializer(Class<?> type, Class<? extends 
Serializer<?>> serializerClass)

3. 自定义TypeInformation 除了使用已有的TypeInformation所定义的数据格式类型之外，用户也可以自定义实现TypeInformation，来满足的不同的数据类型定义需求。Flink提供了可插拔的Type Information Factory让用户将自定义的TypeInformation注册到Flink类型系统中。如下代码所示只需要通过实现org.apache.flink.api.common.typeinfo.TypeInfoFactory接口，返回相应的类型信息。

• 通过@TypeInfo注解创建数据类型，定义CustomTuple数据类型。

@TypeInfo(CustomTypeInfoFactory.class)
public class CustomTuple<T0, T1> {
  public T0 field0;
  public T1 field1;
}

• 然后定义CustomTypeInfoFactory类继承于TypeInfoFactory，参数类型指定CustomTuple。最后重写createTypeInfo方法，创建的CustomTupleTypeInfo就是CustomTuple数据类型TypeInformation。

public class CustomTypeInfoFactory extends TypeInfoFactory<CustomTuple> {
  @Override
  public TypeInformation<CustomTuple> createTypeInfo(Type t, Map<String, TypeInformation<?>> genericParameters) {
    return new CustomTupleTypeInfo(genericParameters.get("T0"), genericParameters.get("T1"));
  }
}