所有的语言开篇都是Hello Word,数据处理引擎也有Hello Word。那就是Word Count。MR,Spark,Flink以来开篇第一个程序都是Word Count。那么今天Flink开始目标就是在本地调试出Word Count。
单机安装Flink
开始Flink之前先在本机尝试安装一下Flink,当然FLink正常情况下是部署的集群方式。作者比较穷,机器配置太低开不了几个虚拟机。所以只能先演示个单机的安装。
Apache Flink需要在Java1.8+以上的环境中运行 。
所以,先确保自己的JDK版本是1.8包含以上的。
Flink单机部署非常简单,只需安装下载安装即可。如果需要与Hadoop版本结合,那么下载相应的Hadoop关联版本即可。如果不与Hadoop结合就直接下载Scala版即可。我这里就直接下载了Scala2.11的相关版本。
点击进入Apache页面进行下载,大小约有283MB。
把下载下来的压缩包进行解压即可。
打开命令行直接执行
/bin/start-cluster.bat
进行启动。
浏览器打开 http://localhost:8081
至此在Windows10环境下即完成Flink的启动。
编写WordCount
因为Flink是由Scala进行开发的,而Scala是基于JVM的一种语言。所以最终也会转换为JAVA字节码文件,所以Flink程序可以由Java、Scala两种语言都可以进行开发。也可以同时开发。比如Java写一部分代码,Scala写另一部分代码。可以参考<Apache Flink利用Maven对Scala与Java进行混编>。
Flink官方提供快速生成工程的两种工具:SBT与Maven。由于作者比较熟悉Maven,( 或者说没用过SBT )。所以直接使用Maven快速创建一个工程。
Java版本
mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.8.0
Scala版本
mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-scala -DarchetypeVersion=1.8.0
按照提示输入相关信息,即可生成最终的项目。
├── pom.xml └── src └── main ├── resources │ └── log4j.properties └── scala/java └── org └── myorg └── quickstart ├── BatchJob.scala └── StreamingJob.scala
把工程导入到IDEA中
如果使用Scala的话,那么需要安装Scala的插件。搜索安装同时需要把Scala语言包进行安装。
不知道如何操作可以联系我 微信公号<指尖数虫>。
package jar; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.ReduceFunction; 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; public class BatchJob { public static void main(String[] args) throws Exception { // set up the batch execution environment final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); //读取目录下的文件 DataSource<String> data = env.readTextFile("/opt/Server_Packets/log/ServerLog_1_runtime.log"); //把文件中的内容按照空格进行拆分为 word,1 1 是为了能够在下面进行计算. data.flatMap(new FlatMapFunction<String, Tuple2<String, Integer() { @Override public void flatMap(String s, Collector<Tuple2<String, Integer collector) throws Exception { for (String word : s.split(" ")){ collector.collect(new Tuple2<>(word,1)); } } }) // 按照元组中的第1位进行分组 .groupBy(0) // 分组的元组的计算方式为 value +value 也就是刚才的 同样的词 把 1+1 .reduce(new ReduceFunction<Tuple2<String, Integer() { @Override public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t2, Tuple2<String, Integer> t1) throws Exception { return new Tuple2<>(t1.f0,t1.f1+ t2.f1); } }) //输出结果 .print(); } }
总结
以上所述是小编给大家介绍的大数据HelloWorld-Flink实现WordCount,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对网站的支持!
如果你觉得本文对你有帮助,欢迎转载,烦请注明出处,谢谢!
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
《魔兽世界》大逃杀!60人新游玩模式《强袭风暴》3月21日上线
暴雪近日发布了《魔兽世界》10.2.6 更新内容,新游玩模式《强袭风暴》即将于3月21 日在亚服上线,届时玩家将前往阿拉希高地展开一场 60 人大逃杀对战。
艾泽拉斯的冒险者已经征服了艾泽拉斯的大地及遥远的彼岸。他们在对抗世界上最致命的敌人时展现出过人的手腕,并且成功阻止终结宇宙等级的威胁。当他们在为即将于《魔兽世界》资料片《地心之战》中来袭的萨拉塔斯势力做战斗准备时,他们还需要在熟悉的阿拉希高地面对一个全新的敌人──那就是彼此。在《巨龙崛起》10.2.6 更新的《强袭风暴》中,玩家将会进入一个全新的海盗主题大逃杀式限时活动,其中包含极高的风险和史诗级的奖励。
《强袭风暴》不是普通的战场,作为一个独立于主游戏之外的活动,玩家可以用大逃杀的风格来体验《魔兽世界》,不分职业、不分装备(除了你在赛局中捡到的),光是技巧和战略的强弱之分就能决定出谁才是能坚持到最后的赢家。本次活动将会开放单人和双人模式,玩家在加入海盗主题的预赛大厅区域前,可以从强袭风暴角色画面新增好友。游玩游戏将可以累计名望轨迹,《巨龙崛起》和《魔兽世界:巫妖王之怒 经典版》的玩家都可以获得奖励。
更新日志
- 崔健《光冻》[首版][WAV+CUE][975M]
- 于台烟.1994-表情【巨石】【WAV+CUE】
- 施文彬.2011-环岛旅行【禾广娱乐】【WAV+CUE】
- 郑俊弘.2016-THE.RED.BOX.LIVE.GREATEST.HITS.2CD【星梦娱乐】【WAV+CUE】
- ABC德国制 《极致HIFI女伶天碟1》6N纯银镀膜[WAV+CUE][1G]
- 刘德华《我和我追逐的梦》台湾首版 [WAV+CUE][859M]
- 林志炫《单身情歌》3CD 超炫精选 [WAV+CUE][1.2G]
- 王嘉文《芳声再艳2》纯银CD[低速原抓WAV+CUE]
- 郭宴《宴乐声色》[低速原抓WAV+CUE]
- 林子祥《爱到发烧SACD(限量版)》[WAV+CUE]
- 群星.1999-爱情麻辣烫电影情歌精丫滚石】【WAV+CUE】
- 许秋怡.1992-偷偷想你【BMG】【WAV+CUE】
- 李龙基.1982-男子汉(2004复刻版)【EMI百代】【WAV+CUE】
- 海来阿木《忧伤歌王DSD》2CD A2HD5纯银镀层 [WAV+CUE][1.5G]
- 林忆莲《呼吸(大碟29)》维京 [WAV+CUE][995M]