原文鏈接:https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/java8.html
Java 8
Java 8引入了一些為編碼的快速性和簡潔性而設(shè)計(jì)的特性，其中最重要的就是我們所說的Lambda表達(dá)式，Java 8 也打開了函數(shù)式編程之門虏辫。Lambda表達(dá)式允許以直接的方式來實(shí)現(xiàn)和傳遞函數(shù)志珍，而無需聲明額外的類。
最新版本的Flink支持所有的Java API操作都可以使用Lambda表達(dá)式钝的，這篇文檔展示了如何使用Lambda表達(dá)式以及描述了一些限制，關(guān)于Flink API的概述，請參考編程指南: https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/api_concepts.html
例子(Examples)
下面的痪寻；例子展示了如何實(shí)現(xiàn)一個(gè)簡單的，內(nèi)部使用Lambda表達(dá)式輸入的map()虽惭。輸入?yún)?shù)i和map函數(shù)的輸出的類型不需要聲明橡类，Java 8的編譯器會(huì)進(jìn)行推斷。
env.fromElements(1, 2, 3)
// returns the squared i
.map(i -> i*i)
.print();
接下來的兩個(gè)例子展示了以Collector作為輸出的兩個(gè)不同的方法實(shí)現(xiàn)芽唇，為了類型安全顾画，函數(shù)如:flatMap()取劫，需要為Collector定義一個(gè)輸出類型(這個(gè)例子中是String)，如果Collector的類型不能通過上下文進(jìn)行推斷的話研侣，就要在Lambda表達(dá)式的參數(shù)列表中手動(dòng)聲明谱邪。不然輸出的結(jié)果將會(huì)被當(dāng)做Object 類型，會(huì)導(dǎo)致非預(yù)期的結(jié)果庶诡。
DataSet<Integer> input = env.fromElements(1, 2, 3);
// collector type must be declared
input.flatMap((Integer number, Collector<String> out) -> {
StringBuilder builder = new StringBuilder();
for(int i = 0; i < number; i++) {
builder.append("a");
out.collect(builder.toString());
}
})
// returns (on separate lines) "a", "a", "aa", "a", "aa", "aaa"
.print();
DataSet<Integer> input = env.fromElements(1, 2, 3);
// collector type must not be declared, it is inferred from the type of the dataset
DataSet<String> manyALetters = input.flatMap((number, out) -> {
StringBuilder builder = new StringBuilder();
for(int i = 0; i < number; i++) {
builder.append("a");
out.collect(builder.toString());
}
});
// returns (on separate lines) "a", "a", "aa", "a", "aa", "aaa"
manyALetters.print();
下面的代碼中顯示了一個(gè)大量使用Lambda表達(dá)式的單詞計(jì)數(shù)器:
DataSet<String> input = env.fromElements("Please count", "the words", "but not this");
// filter out strings that contain "not"
input.filter(line -> !line.contains("not"))
// split each line by space
.map(line -> line.split(" "))
// emit a pair <word,1> for each array element
.flatMap((String[] wordArray, Collector<Tuple2<String, Integer>> out)
-> Arrays.stream(wordArray).forEach(t -> out.collect(new Tuple2<>(t, 1)))
)
// group and sum up
.groupBy(0).sum(1)
// print
.print();
編譯器限制(Compiler Limitations)
當(dāng)前惦银，F(xiàn)link僅支持Eclipse Luna 4.4.2及以上版本的Eclipse JDT 編譯器所編譯的Lambda表達(dá)式。只有Eclipse JDT編譯器保留了對于使用整個(gè)Lambda表達(dá)式特性類型安全的泛型信息末誓。其他的編譯器如: OpenJDK和Oracle JDK的javac扯俱，則扔掉了Lambda表達(dá)式的所有泛型參數(shù)。也就意味著作為Lambda函數(shù)輸入的Tuple2<String, Integer>或者 Collector<String>喇澡，在編譯后的.class文件中會(huì)被裁減成Tuple2或者Collector蘸吓，這對于Flink 編譯器來說信息量太少了。
如何用JDT編譯器編譯一個(gè)包含Lambda表達(dá)式的Flink作業(yè)將在下一章節(jié)介紹撩幽。
使用Eclipse JDT編譯器和Maven編譯Flink任務(wù)
如果你Eclipse IDE的話库继，在做了一些配置之后，你就可以在IDE中運(yùn)行和調(diào)試你的Flink代碼窜醉。Eclipse IDE默認(rèn)使用Eclipse JDT編譯器來編譯Java源代碼宪萄，下一章節(jié)我們將描述如何配置Eclipse IDE。
如果你使用的是其他的IDE如:IntelliJ IDE 或者你想用Maven打jar包來發(fā)布到Flink集群中去執(zhí)行榨惰，你需要修改一下你的項(xiàng)目的pom.xml文件拜英，并使用Maven來編譯你的程序。在快速入門部分中包含了Maven的預(yù)配置文件琅催，你可以在新的項(xiàng)目中使用居凶，或者參考這個(gè)文件。如果你想用Java 8的Lambda表達(dá)式的話藤抡，請取消掉quickstart的pom.xml文件中斜線部分的內(nèi)容侠碧。
注意：你可以手動(dòng)將下面的內(nèi)容插入到Maven的pom.xml文件中，Maven將會(huì)使用Eclipse JDT編譯器來編譯:
<!-- put these lines under "project/build/pluginManagement/plugins" of your pom.xml -->
<plugin>
<!-- Use compiler plugin with tycho as the adapter to the JDT compiler. -->
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.8</source>
<target>1.8</target>
<compilerId>jdt</compilerId>
</configuration>
<dependencies>
<!-- This dependency provides the implementation of compiler "jdt": -->
<dependency>
<groupId>org.eclipse.tycho</groupId>
<artifactId>tycho-compiler-jdt</artifactId>
<version>0.21.0</version>
</dependency>
</dependencies>
</plugin>
如果你是用Eclipse來開發(fā)的話缠黍，m2e插件可能會(huì)與上述插入的內(nèi)容沖突弄兜，導(dǎo)致你的pom.xml非法，如果是這樣的話瓷式，請將下面的內(nèi)容插入到pom.xml中:
<!-- put these lines under "project/build/pluginManagement/plugins/plugin[groupId="org.eclipse.m2e", artifactId="lifecycle-mapping"]/configuration/lifecycleMappingMetadata/pluginExecutions" of your pom.xml -->
<pluginExecution>
<pluginExecutionFilter>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<versionRange>[3.1,)</versionRange>
<goals>
<goal>testCompile</goal>
<goal>compile</goal>
</goals>
</pluginExecutionFilter>
<action>
<ignore></ignore>
</action>
</pluginExecution>
在Eclipse IDE中運(yùn)行和調(diào)試代碼
首先替饿，確保你當(dāng)前運(yùn)行的Eclipse IDE是4.4.2以上的版本，同時(shí)確保你的Eclispe IDE安裝了Java 8 的運(yùn)行環(huán)境贸典。（Window -> Preferences -> Java -> Installed JREs）
創(chuàng)建或者導(dǎo)入你的Eclipse工程
如果你用的是Maven视卢，你需要在你的pom.xml文件中修改maven-compiler-plugin的Java版本。否則右鍵點(diǎn)擊你的工程中的JRE System Library選項(xiàng)廊驼，打開Properties窗口來切換成支持Lambda表達(dá)式的Java 8 JRE据过。
Eclipse JDT編譯器需要指定一個(gè)特殊的編譯器標(biāo)識(shí)來存儲(chǔ).class文件的類型信息颊埃。打開你喜歡的文本編輯器JDT中{project directory}/.setting/org.eclipse.jdt.core.prefs文件，添加下面的內(nèi)容:
org.eclipse.jdt.core.compiler.codegen.lambda.genericSignature=generate
如果不著樣做蝶俱，也可以將下面的Java版本屬性修改成1.8及以上版本:
org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
org.eclipse.jdt.core.compiler.compliance=1.8
org.eclipse.jdt.core.compiler.source=1.8
保存好上面的文件后班利，在Eclipse IDE中刷新整個(gè)項(xiàng)目。
如果你用的是Maven的話榨呆，右鍵Eclipse工程罗标，選擇Maven->update Progect …
如果執(zhí)行下面的代碼無任何異常的話，說明你已經(jīng)配置好了一切
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.fromElements(1, 2, 3).map((in) -> new Tuple1<String>(" " + in)).print();
env.execute();
