【Flink 1.14 实战】Flink DataStream wordcount 快速开始

目录

• • • 快速开始
    • • Maven 快速入门
      • 流处理wordcount
      • 流处理wordcount2
      • 批处理

快速开始

Maven 快速入门

创建项目

唯一的要求是安装Maven 3.0.4（或更高版本）和Java 8.x。

使用以下命令之一创建项目：

$ mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.14.4

构建项目

如果您想构建/打包您的项目，请转到您的项目目录并运行“ mvn clean package”命令。您将找到一个 JAR 文件，其中包含您的应用程序，以及您可能已作为依赖项添加到应用程序的连接器和库：target/-.jar.

流处理wordcount

如下是一个完整的、可运行的程序示例，它是基于流窗口的单词统计应用程序，计算 5 秒窗口内来自 Web 套接字的单词数。你可以复制并粘贴代码以在本地运行。

import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.windowing.time.Time;import org.apache.flink.util.Collector;public class WindowWordCount {    public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Tuple2<String, Integer>> dataStream = env  .socketTextStream("localhost", 9999)  .flatMap(new Splitter())  .keyBy(value -> value.f0)  .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))  .sum(1); dataStream.print(); env.execute("Window WordCount");    }    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> { @Override public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {     for (String word: sentence.split(" ")) {  out.collect(new Tuple2<String, Integer>(word, 1));     } }    }}

修改pom

quick.WindowWordCount

要运行示例程序，首先从终端使用 netcat 启动输入流：

nc -lk 9999

然后，将打包应用程序提交，Flink 的Web UI来提交作业监控集群的状态和正在运行的作业。

只需输入一些单词，然后按回车键即可传入新单词。这些将作为单词统计程序的输入。如果想查看大于 1 的计数，在 5 秒内重复输入相同的单词即可（如果无法快速输入，则可以将窗口大小从 5 秒增加 ☺）。

$ tail log/flink-*-taskexecutor-*.out

流处理wordcount2

import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.api.java.utils.ParameterTool;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.util.Collector;public class WordCount {    public static void main(String[] args) throws Exception { final ParameterTool params = ParameterTool.fromArgs(args); final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.getConfig().setGlobalJobParameters(params); DataStream text; if (params.has("input")) {     text = env.readTextFile(params.get("input")); } else {     System.out.println("Executing WordCount example with default input data set.");     System.out.println("Use --input to specify file input.");     text = env.fromElements("Flink Spark Storm","Flink Flink Flink"); } DataStream<Tuple2> counts = text  .flatMap(new Tokenizer())  .keyBy(value -> value.f0)  .sum(1);    //1 if (params.has("output")) {     counts.writeAsText(params.get("output")); } else {     System.out.println("Printing result to stdout. Use --output to specify output path.");     counts.print(); } env.execute("Streaming WordCount");    }    public static final class Tokenizer implements FlatMapFunction<String, Tuple2> { @Override public void flatMap(String value, Collector<Tuple2> out) {     String[] tokens = value.toLowerCase().split("\\W+");     for (String token : tokens) {  if (token.length() > 0) {      out.collect(new Tuple2(token, 1));  }     } }    }}

修改pom

1.0.1quick.WordCount

然后，将打包应用程序提交，Flink 的Web UI来提交作业监控集群的状态和正在运行的作业。

批处理

 bin/flink run -Dexecution.runtime-mode=BATCH examples/streaming/WordCount.jar

然后，将打包应用程序提交，Flink 的Web UI来提交作业监控集群的状态和正在运行的作业。