需求

詞頻統計，即給一個文件，統計文件中每個單詞出現的次數，分隔符是\t。這個文件內容如下：

hello world welcome hello welcome

統計結果直接打印在控制臺。生產環境下一般Sink到目的地。

使用Flink + java實現需求

環境

JDK：1.8

Maven：3.6.1（最低Maven 3.0.4）

創建項目

mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.8.1 -DarchetypeCatalog=local

groupId: com.vincent artifactId: springboot-flink-train version:1.0 這樣就創建了一個項目，使用Idea導入這個項目，項目結構如下：

里面有兩個自動為我們準備好的java類。

開發步驟

第一步：創建批處理上下文環境

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

第二步：讀取數據

env.readTextFile(textPath);

第三步：transform operations，例如?filter() ?flatMap() ?join() ?coGroup()，這是開發的核心所在，一般就是業務邏輯

第四步：execute program

具體操作

第一步：讀取數據

hello welcome

第二步：每一行的數據按照指定的分隔符拆分

hello welcome

第三步：為每一個單詞賦上次數為1

(hello,1) (welcome,1)

第四步：合并操作

代碼實現

/*** 使用Java API來開發Flink的批處理應用程序*/ public class BatchWCJavaApp {public static void main(String[] args) throws Exception {String input = "E:/test/input/test.txt";// step1: 獲取運行環境ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();// step2: 讀取數據DataSource<String> text = env.readTextFile(input);// step3: transform// FlatMapFunction<String, Tuple2<String, Integer>表示進來一個String, 轉換成一個<String, Integer>類型text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {/**** @param value 就是一行一行的字符串* @param out 轉換成(單詞,次數)* @throws Exception*/@Overridepublic void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {String[] tokens = value.toLowerCase().split("\t");for(String token: tokens) {if(token.length() > 0) {out.collect(new Tuple2<String, Integer>(token, 1));}}}}).groupBy(0).sum(1).print();} }

運行結果

(world,1) (hello,2) (welcome,2)

使用Flink + scala實現需求

環境

JDK：1.8

Maven：3.6.1（最低Maven 3.0.4）

創建項目，跟使用java方式是一樣的

mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-scala -DarchetypeVersion=1.8.1 -DarchetypeCatalog=local

groupId: com.vincent artifactId: springboot-flink-train-scala?version:1.0 這樣就創建了一個項目，使用Idea導入這個項目：

接下來的開發步驟與使用java實現的開發步驟是一樣的：這里給出

代碼實現

import org.apache.flink.api.scala.ExecutionEnvironment/*** 使用Scala開發Flink的批處理應用程序*/ object BatchWCScalaApp {def main(args: Array[String]): Unit = {val input = "E:/test/input/test.txt"val env = ExecutionEnvironment.getExecutionEnvironmentval text = env.readTextFile(input)// 引入隱式轉換import org.apache.flink.api.scala._text.flatMap(_.toLowerCase.split("\t")).filter(_.nonEmpty).map((_, 1)).groupBy(0).sum(1).print()} }

Java與Scala實現方式對比

算子與簡潔性

也就是transform部分雖然原理是一樣的，但是實現的方式不一樣，scala更加簡潔

?

總結

以上是生活随笔為你收集整理的Apache Flink 零基础入门（四）Flink开发批处理应用程序的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。