如何使用 Scala 在 Apache Flink 中执行平均操作

Question

我有一个这样的数据集

15,Rom,36,49
16,Weyoun,22,323
17,Odo,35,13
18,Jean-Luc,45,455

我想选择第3列和第4列作为我的键和值，我如何在Apache Flink中执行平均操作。

我最多能做到“按键分组”。但是我无法对每个键的值执行平均运算。

val lines: DataSet[String] = env.readTextFile("/home/kiran/Desktop/social_friends.csv")

val jn = lines.map(line => line.split(",")).map(word => (word(2).toString,word(3).toInt)).groupBy("0")

Answer 1

这应该有效

val lines: DataSet[String] = env.readTextFile("/home/kiran/Desktop/social_friends.csv")

val jn = lines.map(line => line.split(",")).map(word => (word(2).toString, 1,word(3).toDouble)).groupBy(0).reduce { 
    (left, right) => 
      val (key, left1, left2) = left
      val (_, right1, right2) = right
      (key, left1 + left2, right1 + right2)
}.map(tuple => (tuple._1, tuple._3 / tuple._2))

Answer 2

请注意，我也更改了map。它现在发出3元组：

scala prettyprint-override">val lines: DataSet[String] = env.readTextFile("/home/kiran/Desktop/social_friends.csv")

val jn = lines
  .map(line => line.split(","))
  .map(word => (word(2).toString,word(3).toInt,1))
  .groupBy("0")
  .reduce { (left, right) => (left._1, left._2 + right._2, left._3 + right._3) }
  .map(tuple => (tuple._1, tuple._2 / tuple._3))