上一篇介紹了編寫 Flink 程序的基本步驟，以及一些常見 API，如：map、filter、keyBy 等，重點介紹了 keyBy 方法。本篇將繼續介紹 Flink 中常用的 API，主要內容為：
1.指定 transform 函數
2.Flink 支持的數據類型
3.累加器
1、指定 transform 函數——轉換操作
許多 transform 操作需要用戶自定義函數來實現，Flink 支持多種自定義 transform 函數，接下來一一介紹。
1.1、實現接口
/** * 實現 MapFunction 接口 * 其中泛型的第一 String 代表輸入類型，第二個 Integer 代表輸出類型 */
class MyMapFunction implements MapFunction<String, Integer> { @Override public Integer map(String value) { return Integer.parseInt(value); } } //使用 transform 函數
data.map(new MyMapFunction());
1.2、匿名類
data.map(new MapFunction<String, Integer> () { public Integer map(String value) { return Integer.parseInt(value); } });
匿名類是 Java 語言定義的語法，與 “實現接口” 的方式一樣，只不過不需要顯示定義子類。這種方式比 “實現接口” 更常見一些。
1.3、Java 8 Lambda 表達式
data.map(s -> Integer.parseInt(s)); //或者
data.map(Integer::parseInt);
Java 8 支持 Lambda 表達式，用法與 Scala 語法很像， 寫起來簡潔，並且容易維護，推薦使用這種方式。
1.4、rich function
顧名思義，比普通的 transform 函數要更豐富，額外提供了 4 個方法：open、close、getRuntimeContext 和 setRuntimeContext。它們可以用來創建/初始化本地狀態、訪問廣播變量、訪問累加器和計數器等。感覺有點像 Hadoop 中的 Mapper 或者 Reducer 類。實現上，可以使用自定義類繼承 RichMapFunction 類的方式。
/** * 與實現 MapFunction 接口類似，這里是繼承了 RichMapFunction 類 * 同時可以實現父類更多的方法 */
class MyRichMapFunction extends RichMapFunction<String, Integer> { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); } @Override public RuntimeContext getRuntimeContext() { return super.getRuntimeContext(); } @Override public void setRuntimeContext(RuntimeContext t) { super.setRuntimeContext(t); } @Override public Integer map(String value) throws Exception { return Integer.parseInt(value); } @Override public void close() throws Exception { super.close(); } } data.map(new MyRichMapFunction());
也可以使用匿名類的方式
data.map (new RichMapFunction<String, Integer>() { @Override public void open(Configuration parameters) throws Exception { super.open(parameters); } @Override public RuntimeContext getRuntimeContext() { return super.getRuntimeContext(); } @Override public void setRuntimeContext(RuntimeContext t) { super.setRuntimeContext(t); } @Override public Integer map(String value) { return Integer.parseInt(value); } @Override public void close() throws Exception { super.close(); } });
如果在 rich function 中需要寫較多的業務，那么用匿名類的方式並不簡潔，並且可讀性差。
2、Flink支持的數據類型
目前 Flink 支持 6 種數據類型
Java Tuple 和 Scala Case Class
Java POJO
原子類型
普通類
Values
Hadoop Writable 類型
特殊類
2.1、Java Tuple 和 Scala Case Class
Tuple （元組）是一個混合類型，包含固定數量的屬性，並且每個屬性類型可以不同。例如：二元組有 2 個屬性，類名為 Tuple2；三元組有 3 個屬性，類名為 Tuple3，以此類推。Java 支持的元組為 Tuple1 - Tuple25。訪問屬性可以通過屬性名直接訪問，如：tuple.f4 代表 tuple 的第 5 個屬性。或者使用 tuple.getField(int position) 方法，參數 position 從 0 開始。
/** * Tuple2 二元組作為 DataStream 的輸入類型 */ 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; } } );
2.2、Java POJO
POJO(Plain Ordinary Java Object) 叫做簡單的 Java 對象。滿足以下條件的 Java 或 Scala 類會被 Flink 看做 POJO 類型
類必須是 public
必須有一個 public 修飾的無參構造方法（默認構造器）
所有屬性必須是 public 修飾或者通過 getter 和 setter 方法可以訪問到
屬性類型必須也是 Flink 支持的，Flink 使用 avro 對其序列化
POJO 類型更易使用，且 Flink 更高效地處理 POJO 類型的數據。
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) );
2.3、原子類型
Flink 支持 Java 和 Scala 中所有的原子類型，例如： Integer、String 和 Double 等。
2.4、普通類
不是 POJO 類型的類都會被 Flink 看做是普通的類類型。Flink 將它們視為黑盒且不會訪問它們的內容，普通類類型使用 Kryo 進行序列化與反序列化。這里是第二次提到序列化與反序列化，簡單解釋下這個概念。因為在分布式計算的系統中，不可避免要在不同機器之間傳輸數據，因此為了高效傳輸數據且在不同語言之間互相轉化，需要通過某種協議（protobuf、kryo、avro、json）將對象轉化成另外一種形式（序列化），其他機器接到序列化的數據后再轉化成之前的對象（反序列化）就可以正常使用了。
2.5、Values
不同於一般的序列化框架，Values 類型通過實現 org.apache.flinktypes.Value 接口里的 write 和 read 方法，實現自己的序列化和反序列化邏輯。當一般的序列化框架不夠高效的時候，可以使用 Values 類型。例如：對於一個用數組存儲的稀疏向量。由於數組大多數元素為 0 ，可以僅對非 0 元素進行特殊編碼，而一般的序列化框架會對所有元素進行序列化操作。
Flink 已經預定義了幾種 Value 類型與基本數據類型相對應。如：ByteValue, ShortValue, IntValue, LongValue, FloatValue, DoubleValue, StringValue, CharValue, BooleanValue。這些 Value 類型可以看做是基本數據類型的變體，他們的值是可變的，允許程序重復利用對象，減輕 GC 的壓力。例如：Java 基本數據類型 String 是不可變的，但是 Flink 的 StringValue 類型是可變的。
Flink 定義的 Value 類型與 Hadoop Writable 類型相似，本質都是通過改進基本數據類型的缺點，提供系統整體性能。
2.6、Hadoop Writable
Hadoop Writable 類型也是手動實現了比較高效的序列化與反序列化的邏輯。Value 類型實現了 org.apache.finktypes.Value 接口，而 Hadoop Writable 類型實現了 org.apache.hadoop.Writable 接口，該接口定義了 write 和 readFields 方法用來手動實現序列化與反序列化邏輯。
2.7、特殊類型
特殊類型包括 Scala 中的 Either, Option, and Try 類型，以及 Java API 中的 Either 類型。
3、累加器
累加器可以通過 add 操作，對程序中的某些狀態或者操作進行計數，job 結束后會返回計數的結果。累加器可以用來調試或者記錄信息。
可以自定義累加器，需要實現 Accumulator 接口，當然 Flink 提供了兩種內置的累加器
IntCounter, LongCounter 和 DoubleCounter
Histogram：統計分布
使用累加器的步驟如下：
在 transform 函數中定義累加器對象
private IntCounter numLines = new IntCounter();
注冊累加器對象，可以在 rich function 的 open 方法進行
getRuntimeContext().addAccumulator("num-lines", this.numLines);
在任何需要統計的地方使用累加器
this.numLines.add(1);
獲取累加器結果
myJobExecutionResult.getAccumulatorResult("num-lines")
Job 結束后，累加器的最終值存儲在 JobExecutionResult 對象中，可以通過 execute 方法返回值來獲取 JobExecutionResult 對象。但是對於批處理無法使用調用這個方法（官網沒有提到），可以通過 env.getLastJobExecutionResult 方法獲取。下面是使用累加器的完整示例
public static void main(String[] args) throws Exception { // set up the batch execution environment
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet<String> data = env.readTextFile("你的輸入路徑"); //使用 rich function transform 函數
DataSet<Integer> dataSet = data.map(new MyRichMapFunction()); // 執行程序
dataSet.collect(); // 獲得 job 的結果
JobExecutionResult jobExecutionResult = env.getLastJobExecutionResult(); int res = jobExecutionResult.getAccumulatorResult("num-lines"); // 輸出累機器的值
System.out.println(res); } // 自定義 rich function /** * 與實現 MapFunction 接口類似，這里是繼承了 RichMapFunction 類 * 同時可以實現父類更多的方法 */
class MyRichMapFunction extends RichMapFunction<String, Integer> { /** * 定義累加器 */
private IntCounter numLines = new IntCounter(); @Override public void open(Configuration parameters) throws Exception { // 注冊累加器
getRuntimeContext().addAccumulator("num-lines", this.numLines); } @Override public Integer map(String value) throws Exception { // 累加器自增，記錄處理的行數
this.numLines.add(1); return Integer.parseInt(value); } }
總結
Flink 基本 API 的使用介紹完了，本篇主要介紹了自定義的 transform 函數、Flink 支持的數據類型和累加器。后續會詳細介紹 Flink 的原理、機制以及編程模型。