这篇文章给大家分享的是有关flink batch dataset的示例代码的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。
清原ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景，ssl证书未来市场广阔！成为
成都创新互联公司的ssl证书销售渠道，可以享受市场价格4-6折优惠！如果有意向欢迎电话联系或者加微信：028-86922220（备注：SSL证书合作）期待与您的合作！
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.获得一个执行环境,如果是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的示例代码”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，让大家可以学到更多知识，如果觉得文章不错，可以把它分享出去让更多的人看到吧！
网页标题：flinkbatchdataset的示例代码-创新互联
本文来源：
http://cdkjz.cn/article/cchiho.html
