flink batch dataset的示例代码

这篇文章给大家分享的是有关flink batch dataset的示例代码的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。

package hgs.flink_lessonimport org.apache.flink.api.java.utils.ParameterToolimport org.apache.flink.streaming.api.scala._import org.apache.flink.api.scala.ExecutionEnvironmentimport org.apache.flink.core.fs.FileSystem.WriteModeimport org.apache.flink.api.common.accumulators.Accumulatorimport org.apache.flink.api.common.accumulators.IntCounterimport scala.collection.immutable.Listimport scala.collection.mutable.ListBufferimport scala.collection.immutable.HashMap//import StreamExecutionEnvironment.classobject WordCount {  def main(args: Array[String]): Unit = {     val params = ParameterTool.fromArgs(args)     //1.获得一个执行环境,如果是Streaming则换成StreamExecutionEnvironment     val env = ExecutionEnvironment.getExecutionEnvironment     //这样会得到当前环境下的配置     env.getConfig.setGlobalJobParameters(params)     println(params.get("input"))      println(params.get("output"))     val text = if(params.has("input")){       //2.加载或者创建初始化数据       env.readTextFile(params.get("input"))     }else{       println("Please specify the input file directory.")       return     }          println("lines "+text.count())     val ac =   new IntCounter     //3.在数据上指明操作类型     val counts = text.flatMap{ _.toLowerCase().split("\\W+").filter{_.nonEmpty}}           //这里与spark的算子的groupBy有点不同，这边要用数组类似的下标来确定根据什么进行分组           .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)})      //指明计算后的数据结果放到哪个位置      //4.counts.print()      counts.writeAsCsv("file:/d:/re.txt", "\n", " ",WriteMode.OVERWRITE)      //5.触发程序执行      env.execute("Scala WordCount Example")      //       }}

感谢各位的阅读！关于“flink batch dataset的示例代码”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，让大家可以学到更多知识，如果觉得文章不错，可以把它分享出去让更多的人看到吧！