• Flink程序是实现分布式集合转换的常规程序。集合最初是从源创建的。通过接收器(slink)返回结果，接收器可以将数据写到某个文件或stdout。Flink可以在各种环境(context)中运行，本地JVM或集群。

1.数据集和数据流

• Flink用特殊的类DataSet and DataStream来表示程序中的数据。可以认为他们是可以包含重复数据的不可变数据集合。在DataSet中数据是有限的，而在DataStream中数据是无限的。
• 这些集合不同于java里的集合，他们是不可变的，一旦被创造就不能改动，也不能简单的抽查里面的元素。
• 最初的集合是通过在Flink程序里添加一个源被创造的，新的集合是使用API方法(如map, filter)通过转换得到的。

2.剖析一个Flink程序

• 每个程序包含相同的基本部分：

1. 获得一个执行环境(execution environment).
2. 加载/创建初始数据。
3. 指定转换这些数据。
4. 指定放置计算结果的位置。
5. 触发程序执行。

• StreamExecutionEnvironment是所有Flink程序的基础。可以通过以下静态方法获得：

  getExecutionEnvironment()

  createLocalEnvironment()

  createRemoteEnvironment(String host, int port, String… jarFiles)

  通常只需要使用getExecutionEnvironment()方法,因为这将根据环境做出正确的事：如果你执行你的程序在IDE上或着作为一个普通Java程序，它将创建一个本地环境,将在本地机器上执行程序。如果您从您的程序创建了一个JAR文件，并通过命令行调用它，Flink集群管理者将执行你的main方法并且getExecutionEnvironment()将返回一个在一个集群上执行程序的执行环境。

• 用于指定数据源，执行环境有几个方法来从文件读取：你可以逐行阅读，像CSV文件，或者使用完全自定义数据输入格式。要读取一个文本文件的顺序,您可以使用:

  final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

  DataStream text = env.readTextFile("file:///path/to/file");

  这将给你一个数据流,然后,您可以通过转换创建新的派生数据流。

• 你可以通过调用DataStream数据转换方法转换。例如,一个map转换看起来像这样:

  DataStream input = …;

  DataStream parsed = input.map(new MapFunction() {
  @Override
  public Integer map(String value) {
  return Integer.parseInt(value);
  }
  });

  通过将原始集合中的每个字符串转换为整数，这将创建一个新的数据流。

• 一旦你有了一个包含你的最终结果的数据流，通过创建一个sink可以把它写一个外部系统。这些是一些创建一个sink的的方法示例:

  writeAsText(String path)

  print()

• 一旦你指定完整程序，你需要通过调用StreamExecutionEnvironment上的execute()去触发程序执行。根据ExecutionEnvironment的类型，执行将会被触发在你的本地机器上或提交程序在集群上执行。这个execute()方法返回一个JobExecutionResult,它包含执行时间和累加结果。

3.延迟计算

• 所有Flink程序都是延迟执行的：当一个程序的main方法被执行时，数据加载和转换没有被立刻发生，相反，每个操作被创造，添加到该程序的计划里。当执行明确地被execute() 触发时这些操作被真正执行。

4.指定的keys

• 一些转换(join, coGroup, keyBy, groupBy)要求的集合上定义的一个key。其他的转换(Reduce, GroupReduce, Aggregate, Windows)允许数据在被应用前通过一个key被分组。

• 一个DataSet被分组：

  DataSet<…> input = // […]
  DataSet<…> reduced = input
  .groupBy(/*define key here*/)
  .reduceGroup(/*do something*/);

  一个key在DataStream中被指定：

  DataStream<…> input = // […]
  DataStream<…> windowed = input
  .keyBy(/*define key here*/)
  .window(/*window specification*/);

  Flink的数据模型不是基于 键值( key-value)对的。keys是虚拟的：他们被定义为实际数据的函数来引导分组操作符。

5.指定转换函数

• 许多转换需要用户自定义函数。

• 实现一个接口：

  class MyMapFunction implements MapFunction {
  public Integer map(String value) { return Integer.parseInt(value); }
  };
  data.map(new MyMapFunction());

• 匿名类：

  data.map(new MapFunction () {
  public Integer map(String value) { return Integer.parseInt(value); }
  });

• 需要用户定义函数的所有转换都可以将rich函数作为参数。相比：

  class MyMapFunction implements MapFunction {
  public Integer map(String value) { return Integer.parseInt(value); }
  };

  可以写成：

  class MyMapFunction extends RichMapFunction {
  public Integer map(String value) { return Integer.parseInt(value); }
  };

  像往常一样传递函数到map转换:

  data.map(new MyMapFunction());

  rich函数也可以被定义为一个匿名类：

  data.map (new RichMapFunction() {
  public Integer map(String value) { return Integer.parseInt(value); }
  });

  rich函数除了提供像map一样的用户自定义功能外，还提供四个方法：open, close, getRuntimeContext, and setRuntimeContext。这些是有用的对于向函数传递参数，创建并最终确定本地状态，访问广播变量，和用于访问运行时信息，如累加器和计数器，和信息迭代。

6.支持的数据类型

1. Java Tuples and Scala Case Classes：java元组和Scala case类    元组是包含固定数量的含有各种类型的字段的复合类型。

   DataStream> wordCounts = env.fromElements(
   new Tuple2("hello", 1),
   new Tuple2("world", 2));

   wordCounts.map(new MapFunction, Integer>() {
   @Override
   public Integer map(Tuple2 value) throws Exception {
   return value.f1;
   }
   });

   wordCounts.keyBy(0); // also valid .keyBy("f0")

2. java POJOs：Java和Scala类被Flink作为一种特殊的POJO数据类型对待，并且类必须是public的，它必须有一个没有参数的公共构造函数，所有的字段是公共或必须通过getter和setter函数来访问，对于一个被叫做foo的字段，getter和setter函数必须命名为getFoo()和`setFoo()。`

   public class WordWithCount {

   public String word;  
   public int count;
   
   public WordWithCount() {}
   
   public WordWithCount(String word, int count) {  
       this.word = word;  
       this.count = count;  
   }  

   }

   DataStream wordCounts = env.fromElements(
   new WordWithCount("hello", 1),
   new WordWithCount("world", 2));

   wordCounts.keyBy("word"); // key by field expression "word"

3. Primitive Types   基本数据类型   Flink支持java和Scala基本数据类型。

4. General class types  一般类类型   Flink支持许多Java和Scala类(API和自定义)。所有不确定为POJO类类型(参见上面的POJO需求)是Flink一般类类型。Flink将这些数据类型视为黑盒，无法访问其内容。

5. Value    值类型通过实现org.apache.flinktypes.Value接口的读和写方法为那些操作提供定制代码。当通用串行化是效率无效率的时候使用值类型是合理的。一个例子是一个数据类型实现稀疏向量元素的数组。知道数组大多是零，一个可以用一个特殊的编码输出非零元素,而通用串行化只会写所有的数组元素。    Flink用预定义的值类型,对应于基本数据类型。 (ByteValue, ShortValue, IntValue, LongValue, FloatValue, DoubleValue, StringValue, CharValue, BooleanValue)。这些值类型作为可变的基本数据类型的变量，其价值可以更改，允许程序员重用对象和减轻垃圾收集器的压力。

6. Hadoop Writables

7. Special Types

7.Accumulators & Counters

• Accumulators有简单的结构包括添加操作和工作结束后可用的最后累积的结果。

• 最简单的accumulator是counter：你可以用Accumulator.add(V value)方法使他自己增量。工作结束后Flink将合并所有部分结果并将结果发送给客户端。Accumulators在调试或你想很快知道关于你的数据更多的时候是有用的。

• Flink有以下内置的Accumulators，每一个都实现了Accumulator接口：IntCounter,LongCounter, and IDoubleCounter。

• 如何使用accumulators：                                                                                                                                                        首先你应该你想使用accumulator在用户定义的转换函数处必须创建一个accumulator对象(这里是一个计数器)。

  private IntCounter numLines = new IntCounter();

  第二,你必须注册accumulator对象,通常在rich函数的open()方法。在这里你也可以定义名字。

  getRuntimeContext().addAccumulator("num-lines", this.numLines);

  您现在可以在运算符函数任何地方使用这个accumulator,包括open()和close()方法。

  this.numLines.add(1);

  总的结果将被存储在从执行环境的execute()方法返回的JobExecutionResult对象中。

  myJobExecutionResult.getAccumulatorResult("num-lines")

  每个job中所有的accumulator共享一个命名空间，因此,您可以使用相同的accumulator在你的job的不同运算符函数中。Flink将内部合并(合)并具有相同名称的accumulator。

• 一般的，accumulators的结果只有在这个job结束之后才是可用的。

• Accumulator是最灵活的：它定义了一个类型V用于值的增加，和最终结果类型R用于最终结果。