Windows環(huán)境運(yùn)行FlinkDemo

0.引言

Apache Flink?是一個(gè)分布式的流式計(jì)算引擎，其針對(duì)數(shù)據(jù)流的分布式計(jì)算提供了數(shù)據(jù)分布、數(shù)據(jù)通信以及容錯(cuò)機(jī)制等功能。

我司相關(guān)云服務(wù)產(chǎn)品DLI（數(shù)據(jù)湖探索，Data Lake Insight）完全兼容Apache Flink，且提供了一站式的流處理、批處理、交互式分析的Serverless融合處理分析服務(wù)。本博客，總結(jié)了本人在初入DLI團(tuán)隊(duì)，試著在Windows端編譯Flink官方Demo，并成功提交運(yùn)行FlinkDemo的例子。希望對(duì)有需要了解Apache Flink的新同學(xué)能有所幫助，也可以對(duì)DLI的Flink引擎底層原理有一定認(rèn)識(shí)。

1.Flink安裝

運(yùn)行Flink集群，需要有Java運(yùn)行環(huán)境。

可通過(guò)在?CMD窗口?中運(yùn)行?java -version?驗(yàn)證，若未安裝，需自行提前預(yù)置。

下載Flink v1.9（flink-1.9.0-bin-scala_2.12.tgz），直接解壓即可；

2. 依次進(jìn)入 flink-1.9.0\bin 文件夾，雙擊 start-cluster.bat 文件（此時(shí)會(huì)彈出兩個(gè)java空白窗口）

3. 瀏覽器輸入?http://localhost:8081/#/overview?，正常情況下，會(huì)顯示如下：

2. Demo運(yùn)行

下載的Flink v1.9壓縮包解壓后自帶了幾個(gè)官方Flink示例。在 flink-1.9.0\examples 中可以找到對(duì)應(yīng)的jar包，本博客將演示如何自定義WordCountDemo jar，并運(yùn)行 WordCountDemo。

主類(lèi)WordCount如下（pom.xml和WordCountData.java見(jiàn)附件）：

package org.apache.flink.examples.java.wordcount; 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.api.java.utils.ParameterTool; import org.apache.flink.examples.java.wordcount.util.WordCountData; import org.apache.flink.util.Collector; public class WordCount { public static void main(String[] args) throws Exception { ParameterTool params = ParameterTool.fromArgs(args); ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); env.getConfig().setGlobalJobParameters(params); DataSet text; if (params.has("input")) { text = env.readTextFile(params.get("input")); } else { System.out.println("Executing WordCount example with default input data set."); System.out.println("Use --input to specify file input."); text = WordCountData.getDefaultTextLineDataSet(env); } DataSet> counts = text.flatMap(new Tokenizer()).groupBy(new int[]{0}).sum(1); if (params.has("output")) { counts.writeAsCsv(params.get("output"), "\n", " "); env.execute("WordCount Example"); } else { System.out.println("Printing result to stdout. Use --output to specify output path."); counts.print(); } } public static final class Tokenizer implements FlatMapFunction> { public void flatMap(String value, Collector> out) { String[] tokens = value.toLowerCase().split("\W+"); for (String token : tokens) { if (token.length() > 0) { out.collect(new Tuple2(token, Integer.valueOf(1))); } } } } }

Flink Dashboard提供了Flink任務(wù)的提交功能，按下圖所示將對(duì)應(yīng)的jar包上傳。

算法含義：顧名思義，Word Count會(huì)將輸入文檔中的所有單詞計(jì)數(shù)，這也是大數(shù)據(jù)處理的hello world示例

輸入：任意一個(gè)文本文件

輸出：?jiǎn)卧~和對(duì)應(yīng)的數(shù)量

主類(lèi)：

org.apache.flink.examples.java.wordcount.WordCount

參數(shù)配置：

--input D:\learning\flinkDemo\wordCountInput.txt --output D:\learning\flinkDemo\wordCountRet.txt

執(zhí)行結(jié)果：

附件： wordCountInput.txt 506B 下載次數(shù)：2次

附件： flinkDemo.rar 19.42KB 下載次數(shù)：1次

Flink 數(shù)據(jù)湖探索 DLI

版權(quán)聲明：本文內(nèi)容由網(wǎng)絡(luò)用戶(hù)投稿，版權(quán)歸原作者所有，本站不擁有其著作權(quán)，亦不承擔(dān)相應(yīng)法律責(zé)任。如果您發(fā)現(xiàn)本站中有涉嫌抄襲或描述失實(shí)的內(nèi)容，請(qǐng)聯(lián)系我們jiasou666@gmail.com 處理，核實(shí)后本網(wǎng)站將在24小時(shí)內(nèi)刪除侵權(quán)內(nèi)容。