flink batch dataset的示例代碼

這篇文章給大家分享的是有關(guān)flink batch dataset的示例代碼的內(nèi)容。小編覺得挺實(shí)用的，因此分享給大家做個(gè)參考，一起跟隨小編過來看看吧。

package hgs.flink_lesson

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.core.fs.FileSystem.WriteMode
import org.apache.flink.api.common.accumulators.Accumulator
import org.apache.flink.api.common.accumulators.IntCounter
import scala.collection.immutable.List
import scala.collection.mutable.ListBuffer
import scala.collection.immutable.HashMap

//import StreamExecutionEnvironment.class
object WordCount {
  def main(args: Array[String]): Unit = {
     val params = ParameterTool.fromArgs(args)
     //1.獲得一個(gè)執(zhí)行環(huán)境,如果是Streaming則換成StreamExecutionEnvironment
     val env = ExecutionEnvironment.getExecutionEnvironment
     //這樣會(huì)得到當(dāng)前環(huán)境下的配置
     env.getConfig.setGlobalJobParameters(params)
     println(params.get("input"))
      println(params.get("output"))

     val text = if(params.has("input")){
       //2.加載或者創(chuàng)建初始化數(shù)據(jù)
       env.readTextFile(params.get("input"))
     }else{
       println("Please specify the input file directory.")
       return
     }
     
     println("lines "+text.count())
     val ac =   new IntCounter
     //3.在數(shù)據(jù)上指明操作類型
     val counts = text.flatMap{ _.toLowerCase().split("\\W+").filter{_.nonEmpty}}
           //這里與spark的算子的groupBy有點(diǎn)不同，這邊要用數(shù)組類似的下標(biāo)來確定根據(jù)什么進(jìn)行分組
           .map{(_,1)}.groupBy(0).reduceGroup(it=>{
                                               
                                                
                                               val tuple = it.next()
                                               var cnt = tuple._2 
                                               val ch = tuple._1
                                               while(it.hasNext){
                                                 cnt= cnt+it.next()._2
                                               }
                                               (ch,cnt)})
      //指明計(jì)算后的數(shù)據(jù)結(jié)果放到哪個(gè)位置
      //4.counts.print()
      counts.writeAsCsv("file:/d:/re.txt", "\n", " ",WriteMode.OVERWRITE)
      //5.觸發(fā)程序執(zhí)行
      env.execute("Scala WordCount Example")
      //     

  }
}

感謝各位的閱讀！關(guān)于“flink batch dataset的示例代碼”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，讓大家可以學(xué)到更多知識(shí)，如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到吧！