參考：https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/api_concepts.html

DataSet and DataStream

Anatomy of a Flink Program Flink程式的剖析

• 獲得執行環境， Obtain an execution environment,
• 載入/創建初始數據， Load/create the initial data,
• 指定此數據的轉換， Specify transformations on this data,
• 指定放置計算結果的位置，Specify where to put the results of your computations,
• 觸發程式執行 Trigger the program execution

getExecutionEnvironment()

createLocalEnvironment()

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

通常，你只需要使用getExecutionEnvironment()，因為這將根據上下文執行正確的操作：如果你在IDE中執行程式或作為常規Java程式，它將創建一個本地環境，將執行你的程式 你的本地機器。 如果你從程式中創建了一個JAR文件，並通過命令行調用它，則Flink集群管理器將執行你的main方法，getExecutionEnvironment()將返回一個執行環境，用於在集群上執行你的程式。

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

DataStream<String> input = ...;

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

這將通過將原始集合中的每個String轉換為Integer來創建新的DataStream。

writeAsText(String path)

print()

一旦指定了完整的程式，就需要通過調用StreamExecutionEnvironment上的execute()來觸發程式執行。 根據ExecutionEnvironment的類型，將在本地電腦上觸發執行或提交程式以在群集上執行。

Lazy Evaluation 惰性求值

Specifying Keys 指定鍵

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

DataStream設置鍵：

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

Flink的數據模型不基於鍵值對。 因此，你無需將數據集類型物理打包到鍵和值中。 鍵是“虛擬的”：它們被定義為實際數據上的函數，以指導分組運算符。

Define keys for Tuples 定義元組的鍵

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

元組在第一個欄位（整數類型）上分組。 

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

在這裡，我們將元組分組在由第一個和第二個欄位組成的複合鍵上。

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

指定keyBy(0)將使系統使用完整的Tuple2作為鍵（以Integer和Float為鍵）。 如果要“導航”到嵌套的Tuple2中，則必須使用下麵解釋的欄位表達式鍵。

Define keys using Field Expressions 使用欄位表達式定義鍵

// 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”欄位。
• 按欄位名稱或0偏移欄位索引選擇元組欄位。 例如，“f0”和“5”分別表示Java元組類型的第一和第六欄位。
• 你可以在POJO和Tuples中選擇嵌套欄位。 例如，“user.zip”指的是POJO的“zip”欄位，其存儲在POJO類型的“user”欄位中。 支持任意嵌套和混合POJO和元組，例如“f1.user.zip”或“user.f3.1.zip”。
• 你可以使用“*”通配符表達式選擇完整類型。 這也適用於非Tuple或POJO類型的類型。

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”：選擇Hadoop IntWritable類型。

Define keys using Key Selector Functions 使用鍵選擇器函數定義鍵

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

Specifying Transformation Functions 指定轉換函數

Implementing an interface 實現介面

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

Anonymous classes 匿名類

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

Java 8 Lambdas 表達式

data.filter(s -> s.startsWith("http://"));

data.reduce((i1,i2) -> i1 + i2);

Rich functions 富函數

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); }
};

data.map(new MyMapFunction());

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

Supported Data Types 支持的數據類型

1. 元組（Java Tuples and Scala Case Classes）
2. Java普通對象（Java POJOs）
3. 基本類型（Primitive Types）
4. 常規類（Regular Classes）
5. 值類型（Values）
6. Hadoop可寫介面的實現（Hadoop Writables）
7. 特殊類型（Special Types） 

Tuples and Case Classes 元組

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")

POJO Java普通對象

• 類必須是公共的。
• 它必須有一個沒有參數的公共構造函數（預設構造函數）。
• 所有欄位都是公共的，或者必須通過getter和setter函數訪問。 對於名為foo的欄位，getter和setter方法必須命名為getFoo()和setFoo()。
• Flink必須支持欄位的類型。 目前，Flink使用Avro序列化任意對象（例如Date）。

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 基本類型

General Class Types 常規類類型

Values 值類型

Hadoop Writables Hadoop可寫介面的實現

Special Types 特殊類型

Type Erasure & Type Inference 類型擦除和類型推斷

Accumulators & Counters 累加器和計數器

• IntCounter，LongCounter和DoubleCounter：請參閱下麵的使用計數器的示例。
• 直方圖(Histogram)：離散數量的區間的直方圖實現。 在內部，它只是一個從Integer到Integer的映射。 你可以使用它來計算值的分佈，例如 字數統計程式的每行字數分佈。

How to use accumulators: 如何使用累加器：

private IntCounter numLines = new IntCounter();

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

this.numLines.add(1);

myJobExecutionResult.getAccumulatorResult("num-lines")

Custom accumulators: 定製累加器：