Flink程序會(huì)在分布式的集合上進(jìn)行各類轉(zhuǎn)化操作（如柜某，filter,map,update state,join,group,window,aggregate）嗽元。集合是由數(shù)據(jù)源構(gòu)造出來的（如：文件，kafak topic喂击，或者本地內(nèi)存中的集合等）剂癌。結(jié)果會(huì)返回給sink，sink可能代表著寫入文件或者標(biāo)準(zhǔn)輸出翰绊。Flink程序可以在各種上下文中運(yùn)行佩谷，如standalone,內(nèi)置在其他程序中旁壮。程序執(zhí)行可以發(fā)生在本地JVM，或者集群中的多個(gè)機(jī)器中谐檀。
根據(jù)你的數(shù)據(jù)源是否有界抡谐，你可以選擇使用Dataset API來寫一個(gè)批處理程序或者使用DataStream API來寫一個(gè)流處理程序。這一片文檔會(huì)介紹這兩種API都會(huì)使用到的概念稚补。
DataSet 與 DataStream
Flink使用DataSet 與 DataStream代表程序中的數(shù)據(jù)。你可以將他們看做一個(gè)可以包含重復(fù)元素的不可變集合框喳。DataSet的數(shù)據(jù)時(shí)有限的课幕，而DataSteam的數(shù)據(jù)可以是無限的。
這些集合和普通的java集合在某些方面有不同五垮。首先乍惊，他們是不可變的，意味著一旦它們被創(chuàng)建放仗，不能再添加或移除元素。并且也不可以直接檢查內(nèi)部的元素。
在Flink程序中通過添加source生成集合伯顶，并且可以通過map,filter等API可以生成新的集合害晦。
Flink程序剖析
Flink程序看起來像是一個(gè)轉(zhuǎn)化集合的普通程序。每個(gè)程序都包括下面的部分：
- 獲取執(zhí)行環(huán)境 execution environment
- 加載/創(chuàng)建數(shù)據(jù)
- 在數(shù)據(jù)上定義轉(zhuǎn)化
- 確定計(jì)算結(jié)果輸出到何處
- 觸發(fā)程序計(jì)算
以下代碼均為java代碼
我們將會(huì)對(duì)每個(gè)步驟提供一個(gè)代碼示例惶傻。所有DataSet的核心類可以在 org.apache.flink.api.java 找到棍郎，而DataStream的 API 可以在 org.apache.flink.streaming.api中找到。
StreamExecutionEnvironment是所有Flink程序的基礎(chǔ)银室。你可以通過下面的方式獲韧康琛：
getExecutionEnvironment()
createLocalEnvironment()
createRemoteEnvironment(String host,int port,String... jarFiles)
一般來說，你只需要使用getExecutionEnvironment()即可蜈敢，因?yàn)檫@個(gè)方法會(huì)根據(jù)上下文選擇合適的類：如果你在IDE中執(zhí)行或者作為一個(gè)普通的java程序辜荠，那么這個(gè)方法會(huì)創(chuàng)建一個(gè)本地環(huán)境，你的程序會(huì)在本地機(jī)器運(yùn)行抓狭。如果你將程序打成jar包伯病，并通過命令行提交任務(wù)，F(xiàn)link集群會(huì)執(zhí)行你jar包的main方法否过，getExecutionEnvironment返回的執(zhí)行環(huán)境是集群環(huán)境狱从。
執(zhí)行環(huán)境提供了多種方法從文件中讀取：一行行的讀取叠纹，以csv的格式讀取季研，或者使用自定義的input format讀取文件。你可以如下方式誉察，一行行的讀取內(nèi)容：
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> text = env.readTextFile("file:///path/to/file");
上述步驟會(huì)生成一個(gè)DataStream与涡，你可以在這上面應(yīng)用轉(zhuǎn)化來創(chuàng)建新的DataStream。
下面的幾行直接略過，太簡單驼卖，就是大概介紹了下每個(gè)步驟代碼怎么寫氨肌，給了個(gè)示例。
Lazy Evaluation
所有的flink程序都是懶執(zhí)行的：當(dāng)執(zhí)行程序的main方法時(shí)酌畜，數(shù)據(jù)加載與轉(zhuǎn)化并不會(huì)立即發(fā)生怎囚。而是創(chuàng)建每個(gè)操符并加入程序計(jì)劃plan中。當(dāng)調(diào)用execute()方法時(shí)桥胞，所有的操作符才會(huì)真正執(zhí)行計(jì)算恳守。程序會(huì)在本地執(zhí)行還是在集群執(zhí)行，取決于執(zhí)行環(huán)境的類型贩虾。
lazy evaluation允許用戶構(gòu)造精細(xì)的程序催烘，flink會(huì)將他們看做一個(gè)整體計(jì)劃來執(zhí)行。
指定key
一些轉(zhuǎn)化操作（如join缎罢，coGroup伊群，keyBy，groupBy）要求在數(shù)據(jù)上定義一個(gè)key策精。另一些轉(zhuǎn)化（reduce,groupReduce,aggregate,window）允許數(shù)據(jù)在key上聚合數(shù)據(jù)舰始。
DataStream這樣定義key：
DataStream<...> windowed = input.
.keyBy()
.window();
Flink的數(shù)據(jù)模型不是基于key-value對(duì)的。因此咽袜，你不必一定將數(shù)據(jù)包裹為key-value的形式蔽午。key是“虛擬的”：它被定義為作用在實(shí)際數(shù)據(jù)上的函數(shù)，指導(dǎo)grouping操作符如何定義key酬蹋。
為tuple定義key
最簡單的場景是根據(jù)tuple的field來分組：
DataStream<Tuple3<Integer,String,Long> input = //
KeyedStream<Tuple3<Integer,String,Long>,Tuple> keyed = input.keyBy(0)
tuple根據(jù)第一個(gè)field來分組（Integer類型）
DataStream<Tuple3<Integer,String,Long> input = //
KeyedStream<Tuple3<Integer,String,Long>,Tuple> keyed = input.keyBy(0,1)
上面的示例中及老，我們使用一個(gè)組合key來對(duì)tuple分組，組合key由第一個(gè)和第二個(gè)field組成范抓。
如果在DataStream的數(shù)據(jù)是嵌套結(jié)構(gòu)骄恶，如：
DataStream<Tuple3<Tuple2<Integer, Float>,String,Long>> ds;
定義 keyBy(0)會(huì)導(dǎo)致系統(tǒng)使用整個(gè) Tuple2 作為key。如果你想要使用嵌套結(jié)構(gòu)內(nèi)的field匕垫，可以參照下面的內(nèi)容僧鲁。
使用field表達(dá)式定義key
你可以使用基于字符串的field表達(dá)式來引用嵌套結(jié)構(gòu)內(nèi)的field來定義key，用于group象泵，sort寞秃，join，coGroup等偶惠。field表達(dá)式使得選取復(fù)雜（嵌套）結(jié)構(gòu)如tuple春寿，POJO中的field更容易了。
下面的例子忽孽，我們有一個(gè)WC POJO绑改，它有兩個(gè)field谢床，word與count。想要根據(jù)word來分組厘线，我們可以將field的名字傳遞給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*/);
field表達(dá)式語法：
- 通過field name，選擇 POJO 的field造壮。如 “user” 代表POJO類的userfield
- 通過field name或者從0開始的索引渡讼，選擇 tuple 的field。如 “f0”與“5”分別代表java tuple的第一個(gè)和第6個(gè)field
- 你可以選擇POJO 與 tuple中的嵌套結(jié)構(gòu)中的field耳璧。
- 可以使用“*”通配符成箫，選擇整個(gè)數(shù)據(jù)作為key。這對(duì)于那些不是tuple也不是POJO的數(shù)據(jù)類型很有用
使用key選擇器函數(shù)來定義key
另一個(gè)選擇key的方式是使用key選擇器函數(shù)楞抡。它將數(shù)據(jù)作為入?yún)⑽爸冢祷卦摂?shù)據(jù)的key析藕。key可以是任意類型并且源于確定的計(jì)算召廷。
下面的例子展示了使用key選擇器函數(shù)，返回對(duì)象中的field作為key：
// 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; }
});
指定轉(zhuǎn)化函數(shù)
大多數(shù)轉(zhuǎn)化要求自定義函數(shù)账胧。這一部分列出如何指定自定義函數(shù)的幾種方式竞慢。
實(shí)現(xiàn)接口
最基本的方式是實(shí)現(xiàn)給定的轉(zhuǎn)化接口：
class MyMapFunction implements MapFunction<String, Integer> {
public Integer map(String value) { return Integer.parseInt(value); }
};
data.map(new MyMapFunction());
匿名類
你可以傳入一個(gè)匿名類：
data.map(new MapFunction<String, Integer> () {
public Integer map(String value) { return Integer.parseInt(value); }
});
Java8 Lambda表達(dá)式
flink的java api也支持java 8 的Lambda表達(dá)式：
data.filter(s -> s.startsWith("http://"));
data.reduce((i1,i2) -> i1 + i2);
Rich function
所有要求自定義函數(shù)的轉(zhuǎn)化操作，都可以使用Rich function作為參數(shù)來代替治泥。如筹煮，想要替代：
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); }
};
然后將函數(shù)作為參數(shù)，傳遞給map轉(zhuǎn)化方法中：
data.map(new MyMapFunction());
Rich function 也可以定義成一個(gè)匿名類：
data.map (new RichMapFunction<String, Integer>() {
public Integer map(String value) { return Integer.parseInt(value); }
});
除了用戶需要重寫的方法（map,reduce等）外居夹，Rich function還有四個(gè)方法：open,close,getRuntimeContext與setRuntimeContext败潦。當(dāng)需要傳遞參數(shù)給這些函數(shù)，或者創(chuàng)建以及確定local state准脂，訪問廣播變量 broadcast variable劫扒，訪問運(yùn)行時(shí)信息如accumulator,counter以及獲取iteration的信息等時(shí)，很有用處狸膏。
支持的數(shù)據(jù)類型
Flink對(duì)于什么樣的數(shù)據(jù)類型可以在DataSet或DataStream上處理有一定限制沟饥。這樣做的原因是，flink系統(tǒng)會(huì)分析數(shù)據(jù)類型湾戳，以便選擇一個(gè)高效的執(zhí)行策略贤旷。
Flink支持如6種不同類型的數(shù)據(jù)：（原文確實(shí)是6種）
- Java Tuple 與 Scala Case Class
2.Java POJO - 基本類型
- Regular Class (比POJO 更一般的類）
- Values （實(shí)現(xiàn)了相關(guān)接口的類型）
- Hadoop Writable
- 特殊類型
Tuple 與 Case Class
Tuple（元組）是一個(gè)包含固定數(shù)量，且類型多樣的field的組合類型砾脑。Java API提供了從 Tuple1 到 Tuple25 這25中tuple幼驶。每一個(gè)tuple的field都可以是一個(gè)Flink支持的類型，包括tuple韧衣，這就會(huì)形成嵌套的tuple县遣。tuple的field可以使用field name直接訪問糜颠，如： tuple.f4，或者使用更通用的get方法萧求，如： tuple.getFiled(int position)其兴。field的索引從0開始。注意夸政，這個(gè)規(guī)則與Scala的tuple不同元旬，但是與java的索引方式保持了一致性。
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/Scala的類符合如下要求守问，則在flink中匀归，他們會(huì)被Flink看做是一個(gè)特殊的POJO 數(shù)據(jù)類型：
- calss必須是pulish的
- 必須含有無參構(gòu)造函數(shù)
- 所有的field必須要么是public要么有public才get/set方法。對(duì)于名為 foo 的field耗帕，它的get/set方法必須命名為getFoo() 與 setFoo() （注：這就要求即便是boolean類型穆端，其get/set方法不能再遵循java的習(xí)俗，使用isXXX的命名方式了）
- field的類型必須是flink支持的類型仿便。目前体啰，F(xiàn)link使用Avro系列化任意對(duì)象（如Date）（注：不太明確這個(gè)的意思，是說POJO使用Avro序列化嗽仪？還是說使用Avro序列化那些java提供的非基本類型荒勇，如Date，因?yàn)镈ate不符合這里說的POJO的特點(diǎn)闻坚，起碼field不是pulic的且沒有public的get/set方法）
Flink會(huì)分析POJO的結(jié)構(gòu)沽翔，也就是說，flink會(huì)知道POJO的field有哪些窿凤。因此仅偎，POJO類型比普通的類更容易使用。除此之外雳殊，F(xiàn)link處理POJO類型比普通的類更高效橘沥。
下面給出了一個(gè)示例，一個(gè)簡單的包含兩個(gè)public field的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));
wordCounts.keyBy("word"); // key by field expression "word"
基本類型
Flink支持所有的java/scala基本類型相种，如Integer威恼，String 與 Double。
更普通的類 General Class Type
Flink支持大多數(shù)Java/Scala的類（API自帶的類或自定義的類）寝并。對(duì)于這些類的限制包括：不能被序列化的field箫措，如：file pointer,I/O stream,或其他本地資源(native resource)。遵循java Bean 傳統(tǒng)的類一般來說都可以很好的使用衬潦。
所有沒有符合POJO標(biāo)準(zhǔn)的類會(huì)被Flink任務(wù)是普通類斤蔓。Flink將這些類看做是一個(gè)黑盒子，不能獲取他們的內(nèi)容（如镀岛，為了高效的排序弦牡，應(yīng)該要獲取到他們的content）友驮。這些普通類 General type使用 Kryo 框架來進(jìn)行序列化/反序列化。
Values
Value類型自己就描述了如何進(jìn)行序列化/反序列化驾锰。不同于要經(jīng)過一個(gè)通用的序列化框架來進(jìn)行序列化操作卸留，Value類型通過實(shí)現(xiàn)org.apache.flinktypes.Value接口的 read 與 write方法來定義如何實(shí)現(xiàn)序列化/反序列化操作。當(dāng)使用通用序列化框架會(huì)產(chǎn)生很不高效的情況是椭豫，可以使用Value類型耻瑟。如：一個(gè)包含稀疏矩陣的array作為field的類∩退郑可能array中大部分都是0喳整，因此可以自定義編碼方式來處理這么多的0，而通用序列化框架會(huì)將整個(gè)array的元素都進(jìn)行序列化裸扶。
同理框都，org.apache.flinktypes.CopyableValue接口支持自定義clone的邏輯。
Flink為基本類型自帶了對(duì)應(yīng)的Value接口實(shí)現(xiàn)（ByteValue呵晨，ShortValue, IntValue, LongValue, FloatValue, DoubleValue, StringValue, CharValue, BooleanValue）魏保。這些Value類型可以看做是基本類型變體為值是可變的：他們的值可以被修改，允許程序重用對(duì)象并且減輕垃圾回收的壓力何荚。
Hadoop Writable
你可以使用實(shí)現(xiàn)了org.apache.hadoop.Writable接口的類型囱淋。序列化的邏輯定義在write()中猪杭，readFields()方法用來反序列化（注：感覺原文這里寫錯(cuò)了餐塘，因此翻譯的時(shí)候糾正了過來，原文寫的是readFields方法用來serialization）皂吮。
Special Type 特殊類型
你可以使用一些特殊類型戒傻，包括Scala的Either,Option與Try。Java API也實(shí)現(xiàn)了自己的 Either蜂筹。與Scala的Either類似需纳，它代表著一個(gè)可能是兩種類型之一的值，Left或者Right艺挪。Either在處理錯(cuò)誤或者需要輸出兩種不同類型的結(jié)果時(shí)很有用不翩。
類型擦除與類型推斷
注意：這一部分只存在使用java api場景中
Java編譯器會(huì)在編譯后拋棄掉大部分的泛型信息。這一現(xiàn)象被稱作java中的類型擦除麻裳。這意味著口蝠，在運(yùn)行時(shí)期，一個(gè)對(duì)象的實(shí)例不再清楚的知道它的泛型類型是什么了津坑。如：DataStream<String>的實(shí)例與DataStream<Long>的實(shí)例在JVM中看起來一樣妙蔗。
當(dāng)Flink準(zhǔn)備執(zhí)行程序時(shí)（程序的main方法被調(diào)用時(shí)），會(huì)要求類型信息疆瑰。Flink Java API會(huì)根據(jù)各種方式獲取類型信息（如根據(jù)上游operator的結(jié)果的類型眉反，來判斷當(dāng)前operator的input的類型）并進(jìn)行重構(gòu)昙啄，然后將其存儲(chǔ)在數(shù)據(jù)集與操作符中。你可以使用DataStream.getType()獲取它們寸五。這個(gè)方法會(huì)返回一個(gè)TypeInformation實(shí)例梳凛，TypeInformation是flink內(nèi)部表示類型的一個(gè)類。
類型推斷有它自身的限制梳杏，有時(shí)候需要編程人員的協(xié)助伶跷。如：使用ExecutionEnvironment.fromCollection()方法從集合中創(chuàng)建數(shù)據(jù)集時(shí)，你可以傳入一個(gè)變量秘狞，用于說明數(shù)據(jù)的類型叭莫。另一些含有泛型的方法，如MapFunction<I,O>也有可能需要額外的類型信息烁试。（注：這一部分會(huì)在后面的 Java Lambda Expressions 部分中有詳細(xì)說明）
可以讓input format(自定義數(shù)據(jù)源的接口)與function（轉(zhuǎn)化操作符需要的函數(shù)接口）實(shí)現(xiàn)ResultTypeQueryable接口雇初，確切的告知API它們的返回類型。對(duì)于operator來說减响，所有input的類型一般都可以通過上游的operator的輸出結(jié)果的類型來推斷靖诗。
Accumulator 與 Counter
Accumulator 是一個(gè)由 add 操作符與 final accumulated result組成的簡單的結(jié)構(gòu)。它可以在job結(jié)束后獲取到信息支示。
最簡單的Acumulator是 counter:你可以使用 Accumulator.add(V value)來增加value的值刊橘。job結(jié)束后，F(xiàn)link會(huì)將所有的result累加起來颂鸿，并發(fā)送結(jié)果給client促绵。Accumulator在調(diào)試時(shí)或者你想要快速發(fā)現(xiàn)數(shù)據(jù)更多信息時(shí)很有用。
目前嘴纺，F(xiàn)link有如下內(nèi)置的 累加器败晴。每一個(gè)都實(shí)現(xiàn)了 Acumulator 接口。
- IntCounter,LongCounter與DoubleCounter：下面會(huì)有一個(gè)示例介紹如何使用
- Histogram:histogram直方圖內(nèi)部含有一個(gè)從Integer到Integer的映射栽渴。你可以使用它來計(jì)算離散的值尖坤，如：在字?jǐn)?shù)統(tǒng)計(jì)程序中，計(jì)算每行的字?jǐn)?shù)的分布闲擦。
如何使用累加器：
首先慢味，你必須要在你想要使用的自定義轉(zhuǎn)化function內(nèi)創(chuàng)建一個(gè)accumulator對(duì)象
private IntCounter numLines = new IntCounter();
第二步，你需要注冊這個(gè)累加器對(duì)象墅冷，通常你可以在rich function的open方法中進(jìn)行注冊纯路。這一步，你需要定義累加器的名字
getRuntimeContext().addAccumulator("num-lines", this.numLines);
現(xiàn)在俺榆，你可以在任何地方使用這個(gè)累加器了感昼，包括open和close方法（注：當(dāng)然，僅限這個(gè)自定轉(zhuǎn)化function內(nèi)）
this.numLines.add(1);
全部的結(jié)果罐脊，會(huì)保存在 JobExecutionResult 對(duì)象中定嗓，這個(gè)對(duì)象由 executionEnvironment 調(diào)用execute()方法返回（目前蜕琴，僅當(dāng) execution 會(huì)阻塞直到j(luò)ob結(jié)束的情況下才可用）
myJobExecutionResult.getAccumulatorResult("num-lines")
同一個(gè)job中同一個(gè)名字的累加器會(huì)共享這個(gè)名稱。因此若你想要在不同的function中定義一個(gè)一樣的累加器宵溅，只要將他們的name設(shè)置為一樣即可凌简。Flink會(huì)在內(nèi)部將name相同名稱的累加器的值進(jìn)行merge。
關(guān)于accumulator和iteration的注釋：目前accumulator的結(jié)果僅可在job結(jié)束后才可以獲取恃逻。我們計(jì)劃允許下次iteration能使用之前的iteration的結(jié)果雏搂。你可以使用 Aggregator 來計(jì)算每次迭代統(tǒng)計(jì)信息，并根據(jù)此類統(tǒng)計(jì)信息確定迭代的終止寇损。（注：這一段沒怎么懂）
自定義累加器：
你可以通過實(shí)現(xiàn) Accumulator 接口來實(shí)現(xiàn)自己的累加器凸郑。如果你覺得你的累加器應(yīng)該被推廣到flink，可以創(chuàng)建一個(gè)pull request矛市。
你可以選擇實(shí)現(xiàn) Accumulator 或者 SimpleAccumulator芙沥。
Accumulator<V,R>是最靈活的：它定義了一個(gè)V類型的值作為輸入，最終返回一個(gè)R類型的輸出浊吏。如：直方圖中而昨，V是數(shù)字R是histogram直方圖。SimpleAccumulator 可以用來輸入輸出類型相同的情況找田，如 計(jì)數(shù)器歌憨。
