Apache Flink应用开发之API基本概念(更新中。。。)

API的基本概念

Flink 程序是一种能够对分布式集合进行转换（transformation）的常规程序，比如：过滤、映射、更新状态、联合、分组、定义窗口、聚合等。集合（Collections ）最初是通过源（sources ）来创建（例如：从文件中读取、KAFKA 主题、或者来源于本地，以及从内存中收集）。处理结果是通过槽（sinks）返回的，它可以写入文件（包括分布式文件系统，如HBase），或者直接标准输出，例如命令行终端。Flink程序可以运行在多种环境中，它可以单独运行，或者嵌入到其他程序中，他可以在本地JVM环境中执行，也可以在集群中执行。

依赖于数据源的类型( 有界（bounded） 无界（unbounded）)你可以写一个批处理程序或者流处理程序，DataSet API 是用于批处理程序， DataStream API是用于流处理。这个指南只是介绍两种API的共同部分，关于详细的用法请参考

Streaming Guide 和 Batch Guide

关于实际的例子，我们使用 StreamingExecutionEnvironment 和 DataStream API ，它和DataSet API的概念是一样的，只不过你需要替换成 ExecutionEnvironment 和 DataSet

DataSet and DataStream

Flink 在程序中用DataSet 和 DataStream来标识数据，可以将它们视为可以包含重复的数据的不可变集合，对于DataSet来说数据是有限的，但是对于DataStream，元素的数量是无限的。

这些集合在某些关键地方与普通的Java集合不同。首先，它们是不可变的，这意味着一旦它们被创建，就不能添加或删除元素。也不能轻易的检查内部的元素。

集合最初是通过在代码中添加源(sources)来创建的，并可以通过API中的方法进行转换，譬如 map, filter 等等。

Flink 代码解析

Flink 程序和通常的ETL程序一样看起来没有什么区别，任何的程序都包括下面这些部分：

获取程序执行环境(environment)

加载或者创建初始数据(extract)

定义数据转换(transformations )

定义结果输出(load)

触发并执行程序

note：以Java为例

现在我们大概概括每一个步骤，关于每种数据类型的详细步骤请参考对应的说明文档

DataSet API： org.apache.flink.api.java

DataStream API ：org.apache.flink.streaming.api

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

getExecutionEnvironment()

createLocalEnvironment()

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

通常情况下你可以使用getExecutionEnvironment()这个方法来初始化环境，它会为你做好依赖。如果你是通过IDE或者普通方法创建一个本地环境，那么Flink将在本地执行。或者你可以将你的程序打包成JAR 然后通过命令行在远程集群上执行。

关于具体的数据源读取，方法多种多样，你可以一行一行的读取。或者CSV文件，以及完全自定义的数据格式。要读取一个文本文件作为行序列，您可以使用下面的方法：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

通过上面的方法可以获得一个 DataStream，然后可以应用转换来创建新的派生DataStream

你可以通过在 DataStream 调用转换方法来进行数据转换，举例来说一个map转换：

DataStream<String> input = ...;

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

这将创建一个新的DataStream，将原始集合中的每个字符串转换为整数。一旦你有一个包含最终结果的 DataStream 你就可以通过创建一个 Sink(槽) 把数据写出到外部系统：

writeAsText(String path)

print()

当所有的程序定义完成以后你需要通过调用 StreamExecutionEnvironment 的 execute() 方法来触发程序执行，依据不同的 ExecutionEnvironment 类型触发将在本地或者集群上执行。

execute() 方法将返回一个 JobExecutionResult 结果对象，它包含了执行时间和累加器结果。

跟多的信息请移步 Streaming Guide

Batch Guide

延迟计算

事实上所有的Flink 程序都是延迟执行的，当程序的main方法执行的时候 数据加载和转换不会立即执行，所有的操作都会被添加到执行计划里面，程序是通过在 environment 上明确的调用 execute() 方法来触发执行的，不管程序是在本地执行还是在集群上执行。

定义Keys

一些转换（(join, coGroup, keyBy, groupBy） 需要在集合上定义 key，其他转换(Reduce、groupre引诱、聚合、Windows)允许在应用程序之前将数据分组在一个键上。

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

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

Flink 的数据模型不是基于 key-value 对的，所以不需要物理的设置 key value , 这里的key 是虚拟的，它们被定义在实际的数据之上用来引导 分组操作。

NOTE: In the following discussion we will use the DataStream API and keyBy. For the DataSet API you just have to replace by DataSet and groupBy.

Tuples key的定义

最简单的情
4000
况是将Tuples分组到Tuples的一个或多个字段上

DataStream<Tuple3<Integer,String,Long>> input = // [...]
KeyedStream<Tuple3<Integer,String,Long>,Tuple> keyed = input.keyBy(0)

tuples 分组到第一个字段上

DataStream<Tuple3<Integer,String,Long>> input = // [...]
KeyedStream<Tuple3<Integer,String,Long>,Tuple> keyed = input.keyBy(0,1)

在这里，我们将组放在元组（tuples）第一个字段和第二个字段组成的组合键上

嵌套元组的一个注释:如果您有一个嵌套元组的DataStream，例如:

DataStream<Tuple3<Tuple2<Integer, Float>,String,Long>> ds;

指定keyBy(0)将使系统使用完整的Tuple2作为键(使用整数和浮点数作为键)。如果希望“navigate”到嵌套的Tuple2，则必须使用下面解释的字段表达式键。

利用字段表达式（Field Expressions）定义key

您可以使用基于字符串的字段表达式来引用嵌套的字段，并定义分组、排序、连接或分组的键（ grouping, sorting, joining, or coGrouping）。

字段表达式使得在(嵌套的)复合类型(如Tuple和POJO类型)中选择字段非常容易

在下面的例子中，我们有一个WC POJO，它有两个字段“word”和“count”。根据字段表达式语句，我们只需将其名称传递给keyBy()函数

// some ordinary POJO (Plain old Java Object)
public class WC {
public String word;
public int count;
}
DataStream<WC> words = // [...]
DataStream<WC> wordCounts = words.keyBy("word").window(/*window specification*/);

字段表达式语法

选择POJO中的字段名。例如，“user”指的是POJO类型的“user”字段。

通过字段名的下标偏移量来选择 Tuple，例如，“0”和“5”分别引用了Java Tuple类型的第一个和第6个字段。

原文是这样的，此处应该有书写错误：Select Tuple fields by their field name or 0-offset field index. For example “==f0==” and “5” refer to the first and sixth field of a Java Tuple type, respectively.

可以在pojo和tuple中选择嵌套字段，举例来说：”user.zip” 是 user 这个pojo类的一个字段 “zip”

您可以使用“*”通配符表达式选择full类型。这也适用于不是Tuple或POJO类型的类型。

Field Expression Example:

public static class WC {
public ComplexNestedClass complex; //nested POJO
private int count;
// getter / setter for private field (count)
public int getCount() {
return count;
}
public void setCount(int c) {
this.count = c;
}
}
public static class ComplexNestedClass {
public Integer someNumber;
public float someFloat;
public Tuple3<Long, Long, String> word;
public IntWritable hadoopCitizen;
}

这些是上面的示例代码的有效字段表达式

“count”: WC 类的 count字段

“complex”: 递归地选择POJO类型ComplexNestedClass字段的所有字段。

“complex.word.f2”: 选择嵌套的Tuple3的最后一个字段

“complex.hadoopCitizen”: ComplexNestedClass 类的 hadoopCitizen 字段

通过 Key Selector 定义 key

另一种定义键的方法是“Key Selector”函数,Key Selector函数接受单个元素作为输入，并返回元素的key。key可以是任何类型的，也可以由任意计算派生出来

下面的例子展示了一个键选择器(Key Selector)函数，它简单地返回一个对象的字段:

// some ordinary POJO
public class WC {public String word; public int count;}
DataStream<WC> words = // [...]
KeyedStream<WC> kyed = words
.keyBy(new KeySelector<WC, String>() {
public String getKey(WC wc) { return wc.word; }
});