Apache Flink 基本API概念

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

1.数据集和数据流

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

2.剖析一个Flink程序

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

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

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<String> text = env.readTextFile("file:///path/to/file");
```
这将给你一个数据流,然后,您可以通过转换创建新的派生数据流。

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

DataStream<String> input = ...;

DataStream<Integer> parsed = input.map(new MapFunction<String, Integer>() {
    @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<String, Integer> {
  public Integer map(String value) { return Integer.parseInt(value); }
};
data.map(new MyMapFunction());

匿名类：

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

需要用户定义函数的所有转换都可以将rich函数作为参数。相比：
```
class MyMapFunction implements MapFunction<String, Integer> {
  public Integer map(String value) { return Integer.parseInt(value); }
};
```
可以写成：
```
class MyMapFunction extends RichMapFunction<String, Integer> {
  public Integer map(String value) { return Integer.parseInt(value); }
};
```
像往常一样传递函数到map转换:
```
data.map(new MyMapFunction());
```
rich函数也可以被定义为一个匿名类：
```
data.map (new RichMapFunction<String, Integer>() {
  public Integer map(String value) { return Integer.parseInt(value); }
});
```
rich函数除了提供像map一样的用户自定义功能外，还提供四个方法：open, close, getRuntimeContext, and setRuntimeContext。这些是有用的对于向函数传递参数，创建并最终确定本地状态，访问广播变量，和用于访问运行时信息，如累加器和计数器，和信息迭代。

6.支持的数据类型

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

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

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

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

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<WordWithCount> wordCounts = env.fromElements(
    new WordWithCount("hello", 1),
    new WordWithCount("world", 2));

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

Primitive Types 基本数据类型 Flink支持java和Scala基本数据类型。
General class types 一般类类型 Flink支持许多Java和Scala类(API和自定义)。所有不确定为POJO类类型(参见上面的POJO需求)是Flink一般类类型。Flink将这些数据类型视为黑盒，无法访问其内容。
Value 值类型通过实现org.apache.flinktypes.Value接口的读和写方法为那些操作提供定制代码。当通用串行化是效率无效率的时候使用值类型是合理的。一个例子是一个数据类型实现稀疏向量元素的数组。知道数组大多是零，一个可以用一个特殊的编码输出非零元素,而通用串行化只会写所有的数组元素。 Flink用预定义的值类型,对应于基本数据类型。 (ByteValue, ShortValue, IntValue, LongValue, FloatValue, DoubleValue, StringValue, CharValue, BooleanValue)。这些值类型作为可变的基本数据类型的变量，其价值可以更改，允许程序员重用对象和减轻垃圾收集器的压力。
Hadoop Writables
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>是最灵活的：它定义了一个类型V用于值的增加，和最终结果类型R用于最终结果。