本地安装教程

用几个简单的步骤启动并运行一个Flink示例程序。

安装:下载并启动Flink

Flink可以在Linux、Mac OS X和Windows上运行。要能够运行Flink，惟一的要求是有一个可以工作的Java 8.x安装。Windows用户，请看看Flink on Windows指南，它描述了如何在本地设置的Windows上运行Flink。

你可以发出以下命令，检查Java的正确安装:

  1. java -version

如果您有Java 8，输出将是这样的:

  1. java version "1.8.0_111"
  2. Java(TM) SE Runtime Environment (build 1.8.0_111-b14)
  3. Java HotSpot(TM) 64-Bit Server VM (build 25.111-b14, mixed mode)
  1. 从downloads page下载一个二进制文件。您可以选择任何您喜欢的Hadoop/Scala组合。如果您计划只使用本地文件系统，那么任何Hadoop版本都可以很好地工作。
  2. 转到下载目录。
  3. 解压缩下载的归档文件。
  1. $ cd ~/Downloads        # Go to download directory
  2. $ tar xzf flink-*.tgz   # Unpack the downloaded archive
  3. $ cd flink-1.7.1

MacOS X用户可以通过 Homebrew安装Flink。

  1. $ brew install apache-flink
  2. ...
  3. $ flink --version
  4. Version: 1.2.0, Commit ID: 1c659cf

启动本地Flink集群

  1. $ ./bin/start-cluster.sh  # Start Flink

在http://localhost:8081上检查Dispatcher的web前端，并确保一切正常运行。web前端应该报告一个可用的TaskManager实例。

Dispatcher: Overview

您还可以通过检查logs目录中的日志文件来验证系统是否在运行:

  1. $ tail log/flink-*-standalonesession-*.log
  2. INFO ... - Rest endpoint listening at localhost:8081
  3. INFO ... - http://localhost:8081 was granted leadership ...
  4. INFO ... - Web frontend listening at http://localhost:8081.
  5. INFO ... - Starting RPC endpoint for StandaloneResourceManager at akka://flink/user/resourcemanager .
  6. INFO ... - Starting RPC endpoint for StandaloneDispatcher at akka://flink/user/dispatcher .
  7. INFO ... - ResourceManager akka.tcp://flink@localhost:6123/user/resourcemanager was granted leadership ...
  8. INFO ... - Starting the SlotManager.
  9. INFO ... - Dispatcher akka.tcp://flink@localhost:6123/user/dispatcher was granted leadership ...
  10. INFO ... - Recovering all persisted jobs.
  11. INFO ... - Registering TaskManager ... under ... at the SlotManager.

阅读代码

您可以在scala 和java上找到这个SocketWindowWordCount示例的完整源代码。

  1. object SocketWindowWordCount {
  3.     def main(args: Array[String]) : Unit = {
  5.         // the port to connect to
  6.         val port: Int = try {
  7.             ParameterTool.fromArgs(args).getInt("port")
  8.         } catch {
  9.             case e: Exception => {
  10.                 System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'")
  11.                 return
  12.             }
  13.         }
  15.         // get the execution environment
  16.         val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  18.         // get input data by connecting to the socket
  19.         val text = env.socketTextStream("localhost", port, '\n')
  21.         // parse the data, group it, window it, and aggregate the counts
  22.         val windowCounts = text
  23.             .flatMap { w => w.split("\\s") }
  24.             .map { w => WordWithCount(w, 1) }
  25.             .keyBy("word")
  26.             .timeWindow(Time.seconds(5), Time.seconds(1))
  27.             .sum("count")
  29.         // print the results with a single thread, rather than in parallel
  30.         windowCounts.print().setParallelism(1)
  32.         env.execute("Socket Window WordCount")
  33.     }
  35.     // Data type for words with count
  36.     case class WordWithCount(word: String, count: Long)
  37. }
  1. public class SocketWindowWordCount {
  3.     public static void main(String[] args) throws Exception {
  5.         // the port to connect to
  6.         final int port;
  7.         try {
  8.             final ParameterTool params = ParameterTool.fromArgs(args);
  9.             port = params.getInt("port");
  10.         } catch (Exception e) {
  11.             System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'");
  12.             return;
  13.         }
  15.         // get the execution environment
  16.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         // get input data by connecting to the socket
  19.         DataStream<String> text = env.socketTextStream("localhost", port, "\n");
  21.         // parse the data, group it, window it, and aggregate the counts
  22.         DataStream<WordWithCount> windowCounts = text
  23.             .flatMap(new FlatMapFunction<String, WordWithCount>() {
  24.                 @Override
  25.                 public void flatMap(String value, Collector<WordWithCount> out) {
  26.                     for (String word : value.split("\\s")) {
  27.                         out.collect(new WordWithCount(word, 1L));
  28.                     }
  29.                 }
  30.             })
  31.             .keyBy("word")
  32.             .timeWindow(Time.seconds(5), Time.seconds(1))
  33.             .reduce(new ReduceFunction<WordWithCount>() {
  34.                 @Override
  35.                 public WordWithCount reduce(WordWithCount a, WordWithCount b) {
  36.                     return new WordWithCount(a.word, a.count + b.count);
  37.                 }
  38.             });
  40.         // print the results with a single thread, rather than in parallel
  41.         windowCounts.print().setParallelism(1);
  43.         env.execute("Socket Window WordCount");
  44.     }
  46.     // Data type for words with count
  47.     public static class WordWithCount {
  49.         public String word;
  50.         public long count;
  52.         public WordWithCount() {}
  54.         public WordWithCount(String word, long count) {
  55.             this.word = word;
  56.             this.count = count;
  57.         }
  59.         @Override
  60.         public String toString() {
  61.             return word + " : " + count;
  62.         }
  63.     }
  64. }

运行示例

现在，我们要运行这个Flink应用程序。它将从套接字中读取文本，并且每5秒打印一次前5秒中每个不同单词出现的次数，即一个处理时间的滚动窗口，只要单词是浮动的。

  • 首先，我们使用netcat启动本地服务器
  1. $ nc -l 9000
  • Submit the Flink program:
  1. $ ./bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9000
  2. Starting execution of program

程序连接到套接字并等待输入。您可以检查web界面，以验证作业是否按预期运行:

Dispatcher: Overview (cont'd)Dispatcher: Running Jobs

  • 单词以5秒的时间窗口(处理时间、滚动窗口)计数，并打印为stdout。监视任务管理器的输出文件，并在nc中写入一些文本(单击<return&gt后，一行一行地将输入发送给Flink):</return>
  1. $ nc -l 9000
  2. lorem ipsum
  3. ipsum ipsum ipsum
  4. bye

.outfile将在每次窗口结束时打印计数，只要有单词出现，例如:

  1. $ tail -f log/flink-*-taskexecutor-*.out
  2. lorem : 1
  3. bye : 1
  4. ipsum : 4

要stop Flink当你完成类型:

  1. $ ./bin/stop-cluster.sh

下一个步骤

查看更多的examples，以更好地了解Flink的编程api。完成后，请继续阅读streaming guide。