Flink Java 8文档

原文链接:https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/java8.html
Java 8
Java 8引入了一些为编码的快速性和简洁性而设计的特性，其中最重要的就是我们所说的Lambda表达式，Java 8 也打开了函数式编程之门。Lambda表达式允许以直接的方式来实现和传递函数，而无需声明额外的类。
最新版本的Flink支持所有的Java API操作都可以使用Lambda表达式，这篇文档展示了如何使用Lambda表达式以及描述了一些限制，关于Flink API的概述，请参考编程指南: https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/api_concepts.html
例子(Examples)
下面的；例子展示了如何实现一个简单的，内部使用Lambda表达式输入的map()。输入参数i和map函数的输出的类型不需要声明，Java 8的编译器会进行推断。

env.fromElements(1, 2, 3)
// returns the squared i
.map(i -> i*i)
.print();

接下来的两个例子展示了以Collector作为输出的两个不同的方法实现，为了类型安全，函数如:flatMap()，需要为Collector定义一个输出类型(这个例子中是String)，如果Collector的类型不能通过上下文进行推断的话，就要在Lambda表达式的参数列表中手动声明。不然输出的结果将会被当做Object 类型，会导致非预期的结果。

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

下面的代码中显示了一个大量使用Lambda表达式的单词计数器:

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)
当前，Flink仅支持Eclipse Luna 4.4.2及以上版本的Eclipse JDT 编译器所编译的Lambda表达式。只有Eclipse JDT编译器保留了对于使用整个Lambda表达式特性类型安全的泛型信息。其他的编译器如: OpenJDK和Oracle JDK的javac，则扔掉了Lambda表达式的所有泛型参数。也就意味着作为Lambda函数输入的Tuple2<String, Integer>或者 Collector<String>，在编译后的.class文件中会被裁减成Tuple2或者Collector，这对于Flink 编译器来说信息量太少了。
如何用JDT编译器编译一个包含Lambda表达式的Flink作业将在下一章节介绍。

使用Eclipse JDT编译器和Maven编译Flink任务
如果你Eclipse IDE的话，在做了一些配置之后，你就可以在IDE中运行和调试你的Flink代码。Eclipse IDE默认使用Eclipse JDT编译器来编译Java源代码，下一章节我们将描述如何配置Eclipse IDE。
如果你使用的是其他的IDE如:IntelliJ IDE 或者你想用Maven打jar包来发布到Flink集群中去执行，你需要修改一下你的项目的pom.xml文件，并使用Maven来编译你的程序。在快速入门部分中包含了Maven的预配置文件，你可以在新的项目中使用，或者参考这个文件。如果你想用Java 8的Lambda表达式的话，请取消掉quickstart的pom.xml文件中斜线部分的内容。
注意：你可以手动将下面的内容插入到Maven的pom.xml文件中，Maven将会使用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来开发的话，m2e插件可能会与上述插入的内容冲突，导致你的pom.xml非法，如果是这样的话，请将下面的内容插入到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中运行和调试代码
首先，确保你当前运行的Eclipse IDE是4.4.2以上的版本，同时确保你的Eclispe IDE安装了Java 8 的运行环境。（Window -> Preferences -> Java -> Installed JREs）
创建或者导入你的Eclipse工程
如果你用的是Maven，你需要在你的pom.xml文件中修改maven-compiler-plugin的Java版本。否则右键点击你的工程中的JRE System Library选项，打开Properties窗口来切换成支持Lambda表达式的Java 8 JRE。
Eclipse JDT编译器需要指定一个特殊的编译器标识来存储.class文件的类型信息。打开你喜欢的文本编辑器JDT中{project directory}/.setting/org.eclipse.jdt.core.prefs文件，添加下面的内容:
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中刷新整个项目。
如果你用的是Maven的话，右键Eclipse工程，选择Maven->update Progect …
如果执行下面的代码无任何异常的话，说明你已经配置好了一切

final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
env.fromElements(1, 2, 3).map((in) -> new Tuple1<String>(" " + in)).print();
env.execute();