Aggregate算⼦求最⼤值
gi joepackage comic.ding北京新东方报名
import org.apache.flink.api.java.aggregation.Aggregations
import org.apache.flink.api.scala.{AggregateDataSet, DataSet, ExecutionEnvironment}
import org.apache.flink.streaming.api.scala._陶喆爱很简单英文版
相信用英语怎么说
llection.mutable
object AggregateDemo {
streamlinedef main(args: Array[String]): Unit = {
val env = ExecutionEnvironment
val data = new mutable.MutableList[(Int, String, Double)]
data.+=((1, "yuwen", 89.0))
data.+=((2, "shuxue", 92.2))
学而不思则罔思而不学则殆的翻译data.+=((3, "yuwen", 89.99))
// 构建数据源
人鬼情未了主题曲
val input: DataSet[(Int, String, Double)] = env.fromCollection(data)
val value: AggregateDataSet[(Int, String, Double)] = upBy(1).aggregate(Aggregations.MAX,2)
value.print()
}德语圣诞歌
}
注意：
1，Aggregate只能作⽤于元组上
2，使⽤aggregate，只能使⽤字段索引名或索引名称来进⾏分组 groupBy(0) ，否则会报以下错误:
deluxeedition
Exception in thread "main" java.lang.UnsupportedOperationException: Aggregate does not support grouping with KeySelector functions, yet.
>last but not least
