流處理，這里用netcat來完成

package com.smalltiger.flinkWCimport org.apache.flink.api.java.utils.ParameterTool import org.apache.flink.streaming.api.scala._/*** Created by smalltiger on 2019/11/6.* flink基于流處理的一個WordCount統計*/ object StreamWC {def main(args: Array[String]): Unit = {//從外部命令中獲取參數var params: ParameterTool = ParameterTool.fromArgs(args)var host: String = params.get("host")var port: Int = params.getInt("port")//1.獲取當前執行環境var env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//接受socket文本流val textDstrem:DataStream[String] = env.socketTextStream(host,port)//flatMap和Map需要引入引入隱式轉換import org.apache.flink.api.scala._val dataStream:DataStream[(String,Int)] = textDstrem.flatMap(_.split(" ")).filter(_.nonEmpty).map((_,1)).keyBy(0).sum(1)dataStream.print().setParallelism(1)//啟動executor，執行任務env.execute("啟動任務")} }

批處理，直接處理文件內容

package com.smalltiger.flinkWC import org.apache.flink.api.scala._/*** Created by smalltiger on 2019/11/6.* flink基于批處理統計wordcount*/ object WordCount {def main(args: Array[String]): Unit = {//創建執行環境val env = ExecutionEnvironment.getExecutionEnvironment//從文件中讀取數據val inpath = "D:\\WorkSpace\\flinkWC\\src\\main\\resources\\abc.txt";var inputDS: DataSet[String] = env.readTextFile(inpath)//按照空格進行一個分詞，對單詞進行groupby分組，然后用sum進行一個聚合var wordCounts: AggregateDataSet[(String, Int)] = inputDS.flatMap(_.split(" ")).map((_, 1)).groupBy(0).sum(1)wordCounts.print()} }

總結

以上是生活随笔為你收集整理的flink(一个流处理，一个批处理)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。