站長資訊網(wǎng)一.IDEA開發(fā)環(huán)境
1.pom文件設(shè)置
<properties> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <encoding>UTF-8</encoding> <scala.version>2.11.12</scala.version> <scala.binary.version>2.11</scala.binary.version> <Hadoop.version>2.7.6</hadoop.version> <flink.version>1.6.1</flink.version> </properties> <dependencies> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>${scala.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-scala_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka-0.10_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.38</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.22</version> </dependency> </dependencies> <build> <sourceDirectory>src/main/scala</sourceDirectory> <testSourceDirectory>src/test/scala</testSourceDirectory> <plugins> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <version>3.2.0</version> <executions> <execution> <goals> <goal>compile</goal> <goal>testCompile</goal> </goals> <configuration> <args> <!-- <arg>-make:transitive</arg> --> <arg>-dependencyfile</arg> <arg>${project.build.directory}/.scala_dependencies</arg> </args> </configuration> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-surefire-plugin</artifactId> <version>2.18.1</version> <configuration> <useFile>false</useFile> <disableXmlReport>true</disableXmlReport> <includes> <include>**/*Test.*</include> <include>**/*Suite.*</include> </includes> </configuration> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.0.0</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass>org.apache.spark.WordCount</mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin> </plugins> </build>
2.flink開發(fā)流程
Flink具有特殊類DataSet并DataStream在程序中表示數(shù)據(jù)。您可以將它們視為可以包含重復(fù)項的不可變數(shù)據(jù)集合。在DataSet數(shù)據(jù)有限的情況下，對于一個DataStream元素的數(shù)量可以是無界的。
這些集合在某些關(guān)鍵方面與常規(guī)Java集合不同。首先，它們是不可變的，這意味著一旦創(chuàng)建它們就無法添加或刪除元素。你也不能簡單地檢查里面的元素。
集合最初通過在弗林克程序添加源創(chuàng)建和新的集合從這些通過將它們使用API方法如衍生map，filter等等。
Flink程序看起來像是轉(zhuǎn)換數(shù)據(jù)集合的常規(guī)程序。每個程序包含相同的基本部分：
1.獲取execution environment,
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
2.加載/創(chuàng)建初始化數(shù)據(jù)
DataStream<String> text = env.readTextFile("file:///path/to/file");
3.指定此數(shù)據(jù)的轉(zhuǎn)換
val mapped = input.map { x => x.toInt }
4.指定放置計算結(jié)果的位置
writeAsText(String path)
print()
5.觸發(fā)程序執(zhí)行
在local模式下執(zhí)行程序
execute()
將程序達(dá)成jar運行在線上
./bin/flink run
-m node21:8081
./examples/batch/WordCount.jar
–input hdfs:///user/admin/input/wc.txt
–output hdfs:///user/admin/output2
二. Wordcount案例
1.Scala代碼
package com.xyg.streaming 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 /** * Author: Mr.Deng * Date: 2018/10/15 * Desc: */ object SocketWindowWordCountScala { def main(args: Array[String]) : Unit = { // 定義一個數(shù)據(jù)類型保存單詞出現(xiàn)的次數(shù) case class WordWithCount(word: String, count: Long) // port 表示需要連接的端口 val port: Int = try { ParameterTool.fromArgs(args).getInt("port") } catch { case e: Exception => { System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'") return } } // 獲取運行環(huán)境 val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment // 連接此socket獲取輸入數(shù)據(jù) val text = env.socketTextStream("node21", port, 'n') //需要加上這一行隱式轉(zhuǎn)換 否則在調(diào)用flatmap方法的時候會報錯 import org.apache.flink.api.scala._ // 解析數(shù)據(jù), 分組, 窗口化, 并且聚合求SUM val windowCounts = text .flatMap { w => w.split("\s") } .map { w => WordWithCount(w, 1) } .keyBy("word") .timeWindow(Time.seconds(5), Time.seconds(1)) .sum("count") // 打印輸出并設(shè)置使用一個并行度 windowCounts.print().setParallelism(1) env.execute("Socket Window WordCount") } }
2.Java代碼
package com.xyg.streaming; 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.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.util.Collector; /** * Author: Mr.Deng * Date: 2018/10/15 * Desc: 使用flink對指定窗口內(nèi)的數(shù)據(jù)進(jìn)行實時統(tǒng)計，最終把結(jié)果打印出來 * 先在node21機器上執(zhí)行nc -l 9000 */ public class StreamingWindowWordCountJava { public static void main(String[] args) throws Exception { //定義socket的端口號 int port; try{ ParameterTool parameterTool = ParameterTool.fromArgs(args); port = parameterTool.getInt("port"); }catch (Exception e){ System.err.println("沒有指定port參數(shù)，使用默認(rèn)值9000"); port = 9000; } //獲取運行環(huán)境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //連接socket獲取輸入的數(shù)據(jù) DataStreamSource<String> text = env.socketTextStream("node21", port, "n"); //計算數(shù)據(jù) DataStream<WordWithCount> windowCount = text.flatMap(new FlatMapFunction<String, WordWithCount>() { public void flatMap(String value, Collector<WordWithCount> out) throws Exception { String[] splits = value.split("\s"); for (String word:splits) { out.collect(new WordWithCount(word,1L)); } } })//打平操作，把每行的單詞轉(zhuǎn)為<word,count>類型的數(shù)據(jù) //針對相同的word數(shù)據(jù)進(jìn)行分組 .keyBy("word") //指定計算數(shù)據(jù)的窗口大小和滑動窗口大小 .timeWindow(Time.seconds(2),Time.seconds(1)) .sum("count"); //把數(shù)據(jù)打印到控制臺,使用一個并行度 windowCount.print().setParallelism(1); //注意：因為flink是懶加載的，所以必須調(diào)用execute方法，上面的代碼才會執(zhí)行 env.execute("streaming word count"); } /** * 主要為了存儲單詞以及單詞出現(xiàn)的次數(shù) */ public static class WordWithCount{ public String word; public long count; public WordWithCount(){} public WordWithCount(String word, long count) { this.word = word; this.count = count; } @Override public String toString() { return "WordWithCount{" + "word='" + word + ''' + ", count=" + count + '}'; } } }
3.運行測試
首先，使用nc命令啟動一個本地監(jiān)聽，命令是：
[admin@node21 ~]$ nc -l 9000
通過netstat命令觀察9000端口。 netstat -anlp | grep 9000，啟動監(jiān)聽如果報錯：-bash: nc: command not found，請先安裝nc，在線安裝命令：yum -y install nc。
然后，IDEA上運行flink官方案例程序
node21上輸入
IDEA控制臺輸出如下
4.集群測試
這里單機測試官方案例
[admin@node21 flink-1.6.1]$ pwd /opt/flink-1.6.1 [admin@node21 flink-1.6.1]$ ./bin/start-cluster.sh Starting cluster. Starting standalonesession daemon on host node21. Starting taskexecutor daemon on host node21. [admin@node21 flink-1.6.1]$ jps 2100 StandaloneSessionClusterEntrypoint 2518 TaskManagerRunner 2584 Jps [admin@node21 flink-1.6.1]$ ./bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9000
程序連接到套接字并等待輸入。您可以檢查Web界面以驗證作業(yè)是否按預(yù)期運行：
單詞在5秒的時間窗口（處理時間，翻滾窗口）中計算并打印到stdout。監(jiān)視TaskManager的輸出文件并寫入一些文本nc（輸入在點擊后逐行發(fā)送到Flink）：
三. 使用IDEA開發(fā)離線程序
Dataset是flink的常用程序，數(shù)據(jù)集通過source進(jìn)行初始化，例如讀取文件或者序列化集合，然后通過transformation（filtering、mapping、joining、grouping）將數(shù)據(jù)集轉(zhuǎn)成，然后通過sink進(jìn)行存儲，既可以寫入hdfs這種分布式文件系統(tǒng)，也可以打印控制臺，flink可以有很多種運行方式，如local、flink集群、yarn等.
1. scala程序
package com.xyg.batch import org.apache.flink.api.scala.ExecutionEnvironment import org.apache.flink.api.scala._ /** * Author: Mr.Deng * Date: 2018/10/19 * Desc: */ object WordCountScala{ def main(args: Array[String]) { //初始化環(huán)境 val env = ExecutionEnvironment.getExecutionEnvironment //從字符串中加載數(shù)據(jù) val text = env.fromElements( "Who's there?", "I think I hear them. Stand, ho! Who's there?") //分割字符串、匯總tuple、按照key進(jìn)行分組、統(tǒng)計分組后word個數(shù) val counts = text.flatMap { _.toLowerCase.split("\W+") filter { _.nonEmpty } } .map { (_, 1) } .groupBy(0) .sum(1) //打印 counts.print() } }
2. java程序
package com.xyg.batch; 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.tuple.Tuple2; import org.apache.flink.util.Collector; /** * Author: Mr.Deng * Date: 2018/10/19 * Desc: */ public class WordCountJava { public static void main(String[] args) throws Exception { //構(gòu)建環(huán)境 final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); //通過字符串構(gòu)建數(shù)據(jù)集 DataSet<String> text = env.fromElements( "Who's there?", "I think I hear them. Stand, ho! Who's there?"); //分割字符串、按照key進(jìn)行分組、統(tǒng)計相同的key個數(shù) DataSet<Tuple2<String, Integer>> wordCounts = text .flatMap(new LineSplitter()) .groupBy(0) .sum(1); //打印 wordCounts.print(); } //分割字符串的方法 public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> { @Override public void flatMap(String line, Collector<Tuple2<String, Integer>> out) { for (String word : line.split(" ")) { out.collect(new Tuple2<String, Integer>(word, 1)); } } } }
3.運行
