當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

09_Flink入门案例、word-count程序（java和scala版本）、添加依赖、Flink Streaming和Batch的区别、在集群上执行程序等

發(fā)布時(shí)間：2024/9/27 编程问答 30 豆豆

生活随笔收集整理的這篇文章主要介紹了 09_Flink入门案例、word-count程序（java和scala版本）、添加依赖、Flink Streaming和Batch的区别、在集群上执行程序等小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

1.9.Flink入門案例-wordCount
1.9.1.開發(fā)工具
1.9.2.編寫java版本word-count程序
1.9.2.1.添加Flink Maven依賴
1.9.2.2.編寫wordcount的java代碼
1.9.2.3.數(shù)據(jù)準(zhǔn)備
1.9.2.4.執(zhí)行結(jié)果
1.9.3.編寫scala版本word-count程序
1.9.3.1.添加Flink依賴
1.9.3.2.編寫wordcount的scala程序
1.9.4.Flink StreamingWindowWordCount
1.9.5.Flink程序開發(fā)步驟
1.9.6.Flink Streaming和Batch的區(qū)別 (API使用層面)
1.9.7.在集群上執(zhí)行程序
1.9.8.集群跑jar的時(shí)候pom文件中需要進(jìn)行的build配置

1.9.Flink入門案例-wordCount

1.9.1.開發(fā)工具

?官方建議使用Intellij IDEA，因?yàn)樗J(rèn)集成scala和maven環(huán)境，使用更加方便
?開發(fā)flink程序，可以使用java或者scala語言。
個(gè)人建議，使用scala，因?yàn)閷?shí)現(xiàn)起來更加簡潔。使用java代碼實(shí)現(xiàn)函數(shù)式編程比較別扭。
?建議使用maven國內(nèi)鏡像倉庫地址
國外倉庫下載較慢，可以使用國內(nèi)阿里云的maven倉庫
注意：如果發(fā)現(xiàn)國內(nèi)源下載提示找不到依賴的時(shí)候，記得切換回國外源
國內(nèi)鏡像倉庫配置見備注

1.9.2.編寫java版本word-count程序

新建一個(gè)FLINK IDEA工程,如下：

1.9.2.1.添加Flink Maven依賴

你只要將以下依賴添加到pom.xml中，就能在項(xiàng)目中引入Apache Flink。這些依賴項(xiàng)包含了本地執(zhí)行環(huán)境，因此支持本地測試。
Scala API：為了使用Scala API，將flink-java的artifact id替換為flink-scala_2.11,同時(shí)將flink-streaming-java_2.11替換為flink-streaming-scala_2.11。

總的pom.xml文件內(nèi)容如下：

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.toto.learn</groupId><artifactId>flink-learn</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.11.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.12</artifactId><version>1.11.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.12</artifactId><version>1.11.1</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.6.0</version><configuration><source>1.8</source><target>1.8</target><encoding>UTF-8</encoding></configuration></plugin><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.1.6</version><configuration><scalaCompatVersion>2.11</scalaCompatVersion><scalaVersion>2.11.12</scalaVersion><encoding>UTF-8</encoding></configuration><executions><execution><id>compile-scala</id><phase>compile</phase><goals><goal>add-source</goal><goal>compile</goal></goals></execution><execution><id>test-compile-scala</id><phase>test-compile</phase><goals><goal>add-source</goal><goal>testCompile</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>2.6</version><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs><archive><manifest><mainClass>com.toto.learn.batch.BatchWordCountJava</mainClass></manifest></archive></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build></project>

1.9.2.2.編寫wordcount的java代碼

import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.operators.DataSource; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.util.Collector;/*** @author tuzuoquan* @version 1.0* @ClassName BatchWordCountJava* @description TODO* @date 2020/9/11 14:03**/ public class BatchWordCountJava {public static void main(String[] args) throws Exception {String inputPath = "E:/workspace/wordcount/input";String outputPath = "E:/workspace/wordcount/output";// 獲取運(yùn)行環(huán)境ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();//獲取文件中的內(nèi)容DataSource<String> text = env.readTextFile(inputPath);//groupBy(num) :按照第幾列進(jìn)行排序；sum(num):排序后將第二列的值進(jìn)行求和DataSet<Tuple2<String,Integer>> counts = text.flatMap(new Tokenizer()).groupBy(0).sum(1);counts.writeAsCsv(outputPath,"\n"," ").setParallelism(1);env.execute("batch word count");}public static class Tokenizer implements FlatMapFunction<String, Tuple2<String,Integer>>{@Overridepublic void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {//\w 匹配字母或數(shù)字或下劃線或漢字等價(jià)于[^A-Za-z0-9_]//\W 非數(shù)字字母下劃線String[] tokens = value.toLowerCase().split("\\W+");for (String token: tokens) {if (token.length() > 0) {//轉(zhuǎn)變成 word 1的格式。每個(gè)新的單詞字?jǐn)?shù)都是1out.collect(new Tuple2<String,Integer>(token,1));}}}}}

1.9.2.3.數(shù)據(jù)準(zhǔn)備

words.txt中的內(nèi)容如下：

1.9.2.4.執(zhí)行結(jié)果

output的內(nèi)容如下：

1.9.3.編寫scala版本word-count程序

1.9.3.1.添加Flink依賴

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.toto.learn</groupId><artifactId>flink-learn</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.11.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.12</artifactId><version>1.11.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.12</artifactId><version>1.11.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-scala_2.12</artifactId><version>1.11.1</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.12</artifactId><version>1.11.1</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.6.0</version><configuration><source>1.8</source><target>1.8</target><encoding>UTF-8</encoding></configuration></plugin><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.1.6</version><configuration><scalaCompatVersion>2.11</scalaCompatVersion><scalaVersion>2.11.12</scalaVersion><encoding>UTF-8</encoding></configuration><executions><execution><id>compile-scala</id><phase>compile</phase><goals><goal>add-source</goal><goal>compile</goal></goals></execution><execution><id>test-compile-scala</id><phase>test-compile</phase><goals><goal>add-source</goal><goal>testCompile</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>2.6</version><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs><archive><manifest><mainClass>com.toto.learn.batch.BatchWordCountJava</mainClass></manifest></archive></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build></project>

主要是添加了：

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-scala_2.12</artifactId><version>1.11.1</version> </dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.12</artifactId><version>1.11.1</version> </dependency>

1.9.3.2.編寫wordcount的scala程序

import org.apache.flink.api.scala.ExecutionEnvironmentobject BatchWordCountScala {def main(args: Array[String]): Unit = {val inputPath = "E:/workspace/wordcount/input"val outPut = "E:/workspace/wordcount/output"val env = ExecutionEnvironment.getExecutionEnvironmentval text = env.readTextFile(inputPath)//引入隱式轉(zhuǎn)換import org.apache.flink.api.scala._val counts = text.flatMap(_.toLowerCase.split("\\W")).filter(_.nonEmpty).map((_,1)).groupBy(0).sum(1)counts.writeAsCsv(outPut,"\n"," ").setParallelism(1)env.execute("batch word count")}}

運(yùn)行程序，結(jié)果如上java版本的運(yùn)行結(jié)果一樣

1.9.4.Flink StreamingWindowWordCount

?需求分析
手工通過socket實(shí)時(shí)產(chǎn)生一些單詞，使用flink實(shí)時(shí)接收數(shù)據(jù)，對指定時(shí)間窗口內(nèi)（例如：2秒）的數(shù)據(jù)進(jìn)行聚合統(tǒng)計(jì)，并且把時(shí)間窗口計(jì)算的結(jié)果打印出來。
?代碼開發(fā)
添加對應(yīng)的java依賴或者scala依賴
?執(zhí)行
1：在hadoop上執(zhí)行nc -l 9000
2：在本機(jī)啟動(dòng)idea中的代碼

編寫程序：

import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.util.Collector;/*** @author tuzuoquan* @version 1.0* @ClassName SocketWindowWordCountJava* @description TODO* @date 2020/9/11 17:08**/ public class SocketWindowWordCountJava {public static void main(String[] args) throws Exception {//獲取需要的端口號int port;try {ParameterTool parameterTool = ParameterTool.fromArgs(args);port = parameterTool.getInt("port");}catch (Exception e){System.err.println("No port set. use default port 9000--java");port = 9000;}//1、獲取flink的運(yùn)行環(huán)境org.apache.flink.streaming.api.environment.StreamExecutionEnvironment env = org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment();String hostname = "127.0.0.1";String delimiter = "\n";//2、獲取數(shù)據(jù)源，連接socket獲取輸入的數(shù)據(jù)DataStreamSource<String> text = env.socketTextStream(hostname, port, delimiter);//3、指定算子進(jìn)行運(yùn)算，通過flatMap將數(shù)據(jù)打平DataStream<WordWithCount> windowCounts = text.flatMap(new FlatMapFunction<String, WordWithCount>() {@Overridepublic void flatMap(String value, Collector<WordWithCount> out) throws Exception {String[] splits = value.split("\\s");for (String word : splits) {//默認(rèn)每個(gè)單詞出現(xiàn)1次out.collect(new WordWithCount(word,1L));}}}).keyBy("word") //根據(jù)key進(jìn)行分組，它會(huì)找word中的值.timeWindow(Time.seconds(2),Time.seconds(1)) //指定時(shí)間窗口大小為2秒，指定時(shí)間間隔為1秒.sum("count"); //這里可以使用sum或reduce//4、指定數(shù)據(jù)存儲位置，這里是把它打印到控制臺。把數(shù)據(jù)打印到控制臺并設(shè)置并行度windowCounts.print().setParallelism(1);//5、執(zhí)行，并且給它起個(gè)名字。這一行代碼一定要實(shí)現(xiàn)，否則程序不執(zhí)行。env.execute("Socket window count");}public static class WordWithCount {public String word;public long count;public WordWithCount(){}public WordWithCount(String word,long count){this.word = word;this.count = count;}@Overridepublic String toString() {return "WordWithCount{" +"word='" + word + '\'' +", count=" + count +'}';}} }

Scala的代碼如下：

import org.apache.flink.api.java.utils.ParameterTool import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.streaming.api.windowing.time.Time/*** 滑動(dòng)窗口計(jì)算** 每隔1秒統(tǒng)計(jì)最近2秒內(nèi)的數(shù)據(jù)，打印到控制臺** Created by xxx.xxx on 2018/10/8.*/ object SocketWindowWordCountScala {def main(args: Array[String]): Unit = {//獲取socket端口號val port: Int = try {ParameterTool.fromArgs(args).getInt("port")}catch {case e: Exception => {System.err.println("No port set. use default port 9000--scala")}9000}//獲取運(yùn)行環(huán)境val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//鏈接socket獲取輸入數(shù)據(jù)val text = env.socketTextStream("127.0.0.1",port,'\n')//解析數(shù)據(jù)(把數(shù)據(jù)打平)，分組，窗口計(jì)算，并且聚合求sum//注意：必須要添加這一行隱式轉(zhuǎn)行，否則下面的flatmap方法執(zhí)行會(huì)報(bào)錯(cuò)import org.apache.flink.api.scala._val windowCounts = text.flatMap(line => line.split("\\s"))//打平，把每一行單詞都切開.map(w => WordWithCount(w,1))//把單詞轉(zhuǎn)成word , 1這種形式.keyBy("word")//分組.timeWindow(Time.seconds(2),Time.seconds(1))//指定窗口大小，指定間隔時(shí)間.sum("count");// sum或者reduce都可以//.reduce((a,b)=>WordWithCount(a.word,a.count+b.count))//打印到控制臺windowCounts.print().setParallelism(1);//執(zhí)行任務(wù)env.execute("Socket window count");}case class WordWithCount(word: String,count: Long) }

1.9.5.Flink程序開發(fā)步驟

1：獲得一個(gè)執(zhí)行環(huán)境
2：加載/創(chuàng)建初始化數(shù)據(jù)
3：指定操作數(shù)據(jù)的transaction算子
4：指定把計(jì)算好的數(shù)據(jù)放在哪
5：調(diào)用execute()觸發(fā)執(zhí)行程序
注意：Flink程序是延遲計(jì)算的，只有最后調(diào)用execute()方法的時(shí)候才會(huì)真正觸發(fā)執(zhí)行程序。
延遲計(jì)算好處：你可以開發(fā)復(fù)雜的程序，但是Flink可以將復(fù)雜的程序轉(zhuǎn)成一個(gè)Plan,將Plan作為一個(gè)整體單元執(zhí)行!

1.9.6.Flink Streaming和Batch的區(qū)別 (API使用層面)

?流處理Streaming
StreamExecutionEnvironment
DataStreaming
?批處理Batch
ExecutionEnvironment
DataSet

1.9.7.在集群上執(zhí)行程序

?編譯
?需要在pom文件中添加build配置，打包時(shí)指定入口類全類名【或者在運(yùn)行時(shí)動(dòng)態(tài)指定】
?provided
?mvn clean package
?執(zhí)行
1：在flink機(jī)器上啟動(dòng)local flink集群
2：在flink機(jī)器上執(zhí)行nc -l 9000
3: 在flink機(jī)器上執(zhí)行./bin/flink run -c xxx.xxx.xxx.MainClass FlinkExample-xxxxx.jar --port 9000
4: 在flink機(jī)器上執(zhí)行tail -f log/flink--taskexecutor-.out查看日志輸出
5: 停止任務(wù)
A:web ui界面停止
B:命令執(zhí)行bin/flink cancel

1.9.8.集群跑jar的時(shí)候pom文件中需要進(jìn)行的build配置

<build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.6.0</version><configuration><source>1.8</source><target>1.8</target><encoding>UTF-8</encoding></configuration></plugin><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.1.6</version><configuration><scalaCompatVersion>2.11</scalaCompatVersion><scalaVersion>2.11.12</scalaVersion><encoding>UTF-8</encoding></configuration><executions><execution><id>compile-scala</id><phase>compile</phase><goals><goal>add-source</goal><goal>compile</goal></goals></execution><execution><id>test-compile-scala</id><phase>test-compile</phase><goals><goal>add-source</goal><goal>testCompile</goal></goals></execution></executions></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>2.6</version><configuration><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs><archive><manifest><mainClass>xxx.xxx.SocketWindowWordCountJava</mainClass></manifest></archive></configuration><executions><execution><id>make-assembly</id><phase>package</phase><goals><goal>single</goal></goals></execution></executions></plugin></plugins></build>

總結(jié)

以上是生活随笔為你收集整理的09_Flink入门案例、word-count程序（java和scala版本）、添加依赖、Flink Streaming和Batch的区别、在集群上执行程序等的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇：武夷山旅游 | 一站式吃遍闽北特色小吃！
下一篇： 22_2D函数优化实例