电脑知识|欧美黑人一区二区三区|软件|欧美黑人一级爽快片淫片高清|系统|欧美黑人狂野猛交老妇|数据库|服务器|编程开发|网络运营|知识问答|技术教程文章 - 好吧啦网

您的位置:首頁技術(shù)文章
文章詳情頁

Flink開發(fā)IDEA環(huán)境搭建與測試的方法

瀏覽:176日期:2024-09-02 11:12:04

一.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ù)項(xiàng)的不可變數(shù)據(jù)集合。在DataSet數(shù)據(jù)有限的情況下,對于一個(gè)DataStream元素的數(shù)量可以是無界的。

這些集合在某些關(guān)鍵方面與常規(guī)Java集合不同。首先,它們是不可變的,這意味著一旦創(chuàng)建它們就無法添加或刪除元素。你也不能簡單地檢查里面的元素。

集合最初通過在弗林克程序添加源創(chuàng)建和新的集合從這些通過將它們使用API方法如衍生map,filter等等。

Flink程序看起來像是轉(zhuǎn)換數(shù)據(jù)集合的常規(guī)程序。每個(gè)程序包含相同的基本部分:

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.指定放置計(jì)算結(jié)果的位置

writeAsText(String path)print()

5.觸發(fā)程序執(zhí)行

在local模式下執(zhí)行程序

execute()

將程序達(dá)成jar運(yùn)行在線上

./bin/flink run -m node21:8081 ./examples/batch/WordCount.jar --input hdfs:///user/admin/input/wc.txt--outputhdfs:///user/admin/output2

二.Wordcount案例

1.Scala代碼

package com.xyg.streamingimport org.apache.flink.api.java.utils.ParameterToolimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.streaming.api.windowing.time.Time/** * Author: Mr.Deng * Date: 2018/10/15 * Desc: */object SocketWindowWordCountScala { def main(args: Array[String]) : Unit = { // 定義一個(gè)數(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 } } // 獲取運(yùn)行環(huán)境 val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment // 連接此socket獲取輸入數(shù)據(jù) val text = env.socketTextStream('node21', port, ’n’) //需要加上這一行隱式轉(zhuǎn)換 否則在調(diào)用flatmap方法的時(shí)候會報(bào)錯(cuò) 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è)置使用一個(gè)并行度 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)行實(shí)時(shí)統(tǒng)計(jì),最終把結(jié)果打印出來 * 先在node21機(jī)器上執(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; } //獲取運(yùn)行環(huán)境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //連接socket獲取輸入的數(shù)據(jù) DataStreamSource<String> text = env.socketTextStream('node21', port, 'n'); //計(jì)算數(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') //指定計(jì)算數(shù)據(jù)的窗口大小和滑動窗口大小 .timeWindow(Time.seconds(2),Time.seconds(1)) .sum('count'); //把數(shù)據(jù)打印到控制臺,使用一個(gè)并行度 windowCount.print().setParallelism(1); //注意:因?yàn)閒link是懶加載的,所以必須調(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.運(yùn)行測試

首先,使用nc命令啟動一個(gè)本地監(jiān)聽,命令是:

[admin@node21 ~]$ nc -l 9000

通過netstat命令觀察9000端口。netstat -anlp | grep 9000,啟動監(jiān)聽如果報(bào)錯(cuò):-bash: nc: command not found,請先安裝nc,在線安裝命令:yum -y install nc。

然后,IDEA上運(yùn)行flink官方案例程序

node21上輸入

Flink開發(fā)IDEA環(huán)境搭建與測試的方法

IDEA控制臺輸出如下

Flink開發(fā)IDEA環(huán)境搭建與測試的方法

4.集群測試

這里單機(jī)測試官方案例

[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]$ jpsStandaloneSessionClusterEntrypointTaskManagerRunnerJps[admin@node21 flink-1.6.1]$ ./bin/flink run examples/streaming/SocketWindowWordCount.jar --port 9000

程序連接到套接字并等待輸入。您可以檢查Web界面以驗(yàn)證作業(yè)是否按預(yù)期運(yùn)行:

Flink開發(fā)IDEA環(huán)境搭建與測試的方法

Flink開發(fā)IDEA環(huán)境搭建與測試的方法

單詞在5秒的時(shí)間窗口(處理時(shí)間,翻滾窗口)中計(jì)算并打印到stdout。監(jiān)視TaskManager的輸出文件并寫入一些文本nc(輸入在點(diǎn)擊后逐行發(fā)送到Flink):

Flink開發(fā)IDEA環(huán)境搭建與測試的方法

Flink開發(fā)IDEA環(huán)境搭建與測試的方法

三.使用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可以有很多種運(yùn)行方式,如local、flink集群、yarn等.

1. scala程序

package com.xyg.batchimport org.apache.flink.api.scala.ExecutionEnvironmentimport 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)計(jì)分組后word個(gè)數(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)計(jì)相同的key個(gè)數(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.運(yùn)行

Flink開發(fā)IDEA環(huán)境搭建與測試的方法

到此這篇關(guān)于Flink開發(fā)IDEA環(huán)境搭建與測試的方法的文章就介紹到這了,更多相關(guān)Flink IDEA環(huán)境搭建 內(nèi)容請搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)!

標(biāo)簽: IDEA
相關(guān)文章:
主站蜘蛛池模板: 干培两用箱-细菌恒温培养箱-菲斯福仪器 | 品牌设计_VI设计_电影海报设计_包装设计_LOGO设计-Bacross新越品牌顾问 | 南京交通事故律师-专打交通事故的南京律师 | 【法利莱住人集装箱厂家】—活动集装箱房,集装箱租赁_大品牌,更放心 | 烟台螺纹,烟台H型钢,烟台钢材,烟台角钢-烟台市正丰金属材料有限公司 | DNA亲子鉴定_DNA基因检测中心官方预约平台-严选好基因网 | 比亚迪叉车-比亚迪电动叉车堆垛车托盘车仓储叉车价格多少钱报价 磁力去毛刺机_去毛刺磁力抛光机_磁力光饰机_磁力滚抛机_精密金属零件去毛刺机厂家-冠古科技 | 双菱电缆-广州电缆厂_广州电缆厂有限公司 | 模温机-油温机-电加热导热油炉-工业冷水机「欧诺智能」 | 仪器仪表网 - 永久免费的b2b电子商务平台 | WTB5光栅尺-JIE WILL磁栅尺-B60数显表-常州中崴机电科技有限公司 | WF2户外三防照明配电箱-BXD8050防爆防腐配电箱-浙江沃川防爆电气有限公司 | 无线讲解器-导游讲解器-自助讲解器-分区讲解系统 品牌生产厂家[鹰米讲解-合肥市徽马信息科技有限公司] | 精密五金冲压件_深圳五金冲压厂_钣金加工厂_五金模具加工-诚瑞丰科技股份有限公司 | 胜为光纤光缆_光纤跳线_单模尾纤_光纤收发器_ODF光纤配线架厂家直销_北京睿创胜为科技有限公司 - 北京睿创胜为科技有限公司 | 东莞注册公司-代办营业执照-东莞公司注册代理记账-极刻财税 | 单螺旋速冻机-双螺旋-流态化-隧道式-食品速冻机厂家-广州冰泉制冷 | 猎头招聘_深圳猎头公司_知名猎头公司| 厂房出租_厂房出售_产业园区招商_工业地产&nbsp;-&nbsp;中工招商网 | 北京公寓出租网-北京酒店式公寓出租平台 | 磁力加热搅拌器-多工位|大功率|数显恒温磁力搅拌器-司乐仪器官网 | 福州时代广告制作装饰有限公司-福州广告公司广告牌制作,福州展厅文化墙广告设计, | 蔬菜清洗机_环速洗菜机_异物去除清洗机_蔬菜清洗机_商用洗菜机 - 环速科技有限公司 | 低温柔性试验仪-土工布淤堵-沥青车辙试验仪-莱博特(天津)试验机有限公司 | bng防爆挠性连接管-定做金属防爆挠性管-依客思防爆科技 | 破碎机_上海破碎机_破碎机设备_破碎机厂家-上海山卓重工机械有限公司 | 冲击式破碎机-冲击式制砂机-移动碎石机厂家_青州市富康机械有限公司 | 高速混合机_锂电混合机_VC高效混合机-无锡鑫海干燥粉体设备有限公司 | 书信之家_书信标准模板范文大全 最新范文网_实用的精品范文美文网 | 能量回馈_制动单元_电梯节能_能耗制动_深圳市合兴加能科技有限公司 | 上海办公室装修公司_办公室设计_直营办公装修-羚志悦装 | 土壤检测仪器_行星式球磨仪_土壤团粒分析仪厂家_山东莱恩德智能科技有限公司 | 粉碎机_塑料粉碎机_塑料破碎机厂家-星标机械 | 仓储笼_金属箱租赁_循环包装_铁网箱_蝴蝶笼租赁_酷龙仓储笼租赁 测试治具|过炉治具|过锡炉治具|工装夹具|测试夹具|允睿自动化设备 | 瑞典Blueair空气净化器租赁服务中心-专注新装修办公室除醛去异味服务! | 周口市风机厂,周鼓风机,河南省周口市风机厂| 广州/东莞小字符喷码机-热转印打码机-喷码机厂家-广州瑞润科技 | 酵素生产厂家_酵素OEM_酵素加盟_酵素ODM_酵素原料厂家_厦门益力康 | 气动|电动调节阀|球阀|蝶阀-自力式调节阀-上海渠工阀门管道工程有限公司 | 小型数控车床-数控车床厂家-双头数控车床 | 小型手持气象站-空气负氧离子监测站-多要素微气象传感器-山东天合环境科技有限公司 |