Sample Project using the Java API

要求
创建工程
检查工程
编译工程
下一步

通过简单地几步来开始编写你的 Flink Java 程序。

要求

唯一的要求是需要安装 Maven 3.0.4 (或者更高)和 Java 7.x (或者更高)

创建工程

使用下面其中一个命令来创建Flink Java工程

Use Maven archetypes
Run the quickstart script

$ mvn archetype:generate                               \
      -DarchetypeGroupId=org.apache.flink              \
      -DarchetypeArtifactId=flink-quickstart-java      \
      -DarchetypeCatalog=https://repository.apache.org/content/repositories/snapshots/ \
      -DarchetypeVersion=1.3-SNAPSHOT

这种方式允许你为新创建的工程命名。而且会以交互式地方式询问你为 groupId, artifactId 以及 package 命名。

$ curl https://flink.apache.org/q/quickstart-SNAPSHOT.sh | bash

检查工程

运行完上面的命令会在当前工作目录下创建一个新目录。如果你使用了 curl 命令来创建 Flink Java 工程，这个目录的名称是 quickstart。否则，就是你输入的 artifactId 名字：

$ tree quickstart/
quickstart/
├── pom.xml
└── src
    └── main
        ├── java
        │   └── org
        │       └── myorg
        │           └── quickstart
        │               ├── BatchJob.java
        │               ├── SocketTextStreamWordCount.java
        │               ├── StreamingJob.java
        │               └── WordCount.java
        └── resources
            └── log4j.properties

这个工程是一个 Maven 工程, 包含四个类。 StreamingJob 和 BatchJob 是基本的框架程序，SocketTextStreamWordCount 是一个 Streaming 示例；而 WordCount 是一个 Batch 示例。需要注意的是，所有这些类的 main 方法都允许你在开发/测试模式下启动Flink。

我们推荐将这个工程导入到你的IDE中，并进行开发和测试。如果你用的是 Eclipse，可以使用 m2e 插件来导入 Maven 工程。有些Eclipse发行版默认嵌入了这个插件，其他的需要你手动去安装。IntelliJ IDE内置就提供了对 Maven 工程的支持。

给Mac OS X用户的建议：默认的 JVM 堆内存对 Flink 来说太小了，你必须手动增加内存。这里以 Eclipse 为例，依次选择 Run Configurations -> Arguments，然后在 VM Arguments 里写入：-Xmx800m。

编译工程

如果你想要 编译你的工程 , 进入到工程所在目录，并输入 mvn clean install -Pbuild-jar 命令。你将会找到 target/original-your-artifact-id-your-version.jar 文件，它可以在任意的 Flink 集群上运行。还有一个 fat-jar，名为 target/your-artifact-id-your-version.jar ，包含了所有添加到 Maven 工程的依赖。

下一步

编写我们自己的程序！

Quickstart 工程包含了一个 WordCount 的实现，也就是大数据处理系统的 Hello World。WordCount 的目标是计算文本中单词出现的频率。比如：单词 “the” 或者 “house” 在所有的Wikipedia文本中出现了多少次。

样本输入:

big data is big

样本输出:

big 2
data 1
is 1

下面的代码就是 Quickstart 工程的 WordCount 实现，它使用两种操作( FlatMap 和 Reduce )处理了一些文本，并且在标准输出中打印了单词的计数结果。

public class WordCount {

  public static void main(String[] args) throws Exception {

    // set up the execution environment
    final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

    // get input data
    DataSet<String> text = env.fromElements(
        "To be, or not to be,--that is the question:--",
        "Whether 'tis nobler in the mind to suffer",
        "The slings and arrows of outrageous fortune",
        "Or to take arms against a sea of troubles,"
        );

    DataSet<Tuple2<String, Integer>> counts =
        // split up the lines in pairs (2-tuples) containing: (word,1)
        text.flatMap(new LineSplitter())
        // group by the tuple field "0" and sum up tuple field "1"
        .groupBy(0)
        .sum(1);

    // execute and print result
    counts.print();
  }
}

这些操作是在专门的类中定义的，下面是 LineSplitter 类的实现。

public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {

  @Override
  public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
    // normalize and split the line
    String[] tokens = value.toLowerCase().split("\\W+");

    // emit the pairs
    for (String token : tokens) {
      if (token.length() > 0) {
        out.collect(new Tuple2<String, Integer>(token, 1));
      }
    }
  }
}

完整代码参见 Check GitHub。

有关 API 的完整概述，请参见 DataStream API 和 DataSet API 章节。如果你遇到困难，请到邮件列表里面提问。我们很乐意为你解答。