所有的語(yǔ)言開(kāi)篇都是Hello Word,數(shù)據(jù)處理引擎也有Hello Word。那就是Word Count。MR，Spark，F(xiàn)link以來(lái)開(kāi)篇第一個(gè)程序都是Word Count。那么今天Flink開(kāi)始目標(biāo)就是在本地調(diào)試出Word Count。
單機(jī)安裝Flink
開(kāi)始Flink之前先在本機(jī)嘗試安裝一下Flink,當(dāng)然FLink正常情況下是部署的集群方式。作者比較窮，機(jī)器配置太低開(kāi)不了幾個(gè)虛擬機(jī)。所以只能先演示個(gè)單機(jī)的安裝。
Apache Flink需要在Java1.8+以上的環(huán)境中運(yùn)行 。
所以,先確保自己的JDK版本是1.8包含以上的。
Flink單機(jī)部署非常簡(jiǎn)單，只需安裝下載安裝即可。如果需要與Hadoop版本結(jié)合，那么下載相應(yīng)的Hadoop關(guān)聯(lián)版本即可。如果不與Hadoop結(jié)合就直接下載Scala版即可。我這里就直接下載了Scala2.11的相關(guān)版本。
點(diǎn)擊進(jìn)入Apache頁(yè)面進(jìn)行下載，大小約有283MB。
把下載下來(lái)的壓縮包進(jìn)行解壓即可。
打開(kāi)命令行直接執(zhí)行
/bin/start-cluster.bat
進(jìn)行啟動(dòng)。
瀏覽器打開(kāi) http://localhost:8081
至此在Windows10環(huán)境下即完成Flink的啟動(dòng)。
編寫WordCount
因?yàn)镕link是由Scala進(jìn)行開(kāi)發(fā)的，而Scala是基于JVM的一種語(yǔ)言。所以最終也會(huì)轉(zhuǎn)換為JAVA字節(jié)碼文件，所以Flink程序可以由Java、Scala兩種語(yǔ)言都可以進(jìn)行開(kāi)發(fā)。也可以同時(shí)開(kāi)發(fā)。比如Java寫一部分代碼，Scala寫另一部分代碼。可以參考Apache Flink利用Maven對(duì)Scala與Java進(jìn)行混編>。
Flink官方提供快速生成工程的兩種工具：SBT與Maven。由于作者比較熟悉Maven，( 或者說(shuō)沒(méi)用過(guò)SBT )。所以直接使用Maven快速創(chuàng)建一個(gè)工程。
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
按照提示輸入相關(guān)信息,即可生成最終的項(xiàng)目。
├── pom.xml
└── src
└── main
├── resources
│ └── log4j.properties
└── scala/java
└── org
└── myorg
└── quickstart
├── BatchJob.scala
└── StreamingJob.scala
把工程導(dǎo)入到IDEA中
如果使用Scala的話，那么需要安裝Scala的插件。搜索安裝同時(shí)需要把Scala語(yǔ)言包進(jìn)行安裝。
不知道如何操作可以聯(lián)系我 微信公號(hào)指尖數(shù)蟲(chóng)>。
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();
//讀取目錄下的文件
DataSourceString> data = env.readTextFile("/opt/Server_Packets/log/ServerLog_1_runtime.log");
//把文件中的內(nèi)容按照空格進(jìn)行拆分為 word,1 1 是為了能夠在下面進(jìn)行計(jì)算.
data.flatMap(new FlatMapFunctionString, Tuple2String, Integer>>() {
@Override
public void flatMap(String s, CollectorTuple2String, Integer>> collector) throws Exception {
for (String word : s.split(" ")){
collector.collect(new Tuple2>(word,1));
}
}
})
// 按照元組中的第1位進(jìn)行分組
.groupBy(0)
// 分組的元組的計(jì)算方式為 value +value 也就是剛才的 同樣的詞 把 1+1
.reduce(new ReduceFunctionTuple2String, Integer>>() {
@Override
public Tuple2String, Integer> reduce(Tuple2String, Integer> t2, Tuple2String, Integer> t1) throws Exception {
return new Tuple2>(t1.f0,t1.f1+ t2.f1);
}
})
//輸出結(jié)果
.print();
}
}
總結(jié)
以上所述是小編給大家介紹的大數(shù)據(jù)HelloWorld-Flink實(shí)現(xiàn)WordCount,希望對(duì)大家有所幫助，如果大家有任何疑問(wèn)請(qǐng)給我留言，小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持！
如果你覺(jué)得本文對(duì)你有幫助，歡迎轉(zhuǎn)載，煩請(qǐng)注明出處，謝謝！
您可能感興趣的文章:- 淺談實(shí)時(shí)計(jì)算框架Flink集群搭建與運(yùn)行機(jī)制
- 詳解大數(shù)據(jù)處理引擎Flink內(nèi)存管理
- Apache FlinkCEP 實(shí)現(xiàn)超時(shí)狀態(tài)監(jiān)控的步驟詳解
- Flink支持哪些數(shù)據(jù)類型?
- Java lambda表達(dá)式實(shí)現(xiàn)Flink WordCount過(guò)程解析
- 解析Flink內(nèi)核原理與實(shí)現(xiàn)核心抽象
