Scala Spark - 丢弃空键

Question

我有以下 map :

 val pairs = lines.map( l => ( if (l.split(",")(1).toInt < 60) { "rest" } else if (l.split(",")(1).toInt > 110) { "sport" }, 10) ).reduceByKeyAndWindow((a:Int, b:Int) => (a+b), Seconds(12))

基本上，当某人的 HR 低于 60 时，它被归类为休息，高于 110 被归类为运动。元组的第二个变量表示此人已经做了 10 分钟。

现在，这为 60 到 110 之间的值映射了一个空键。我想要的是完全丢弃它们。如何实现？

于是从

("rest", 30)
("sport", 120)
((),10)

我正在尝试过滤掉 ((),10)。我试过了

 pairs.filter{case (key, value) => key.length < 3} //error: value length is not a member of Any
 pairs.filter(_._1 != "")  //no error, just still keeps the empty keys, too   

似乎都不起作用。

Answer 1

您的问题是您的 if 表达式在匹配 Unit 的情况下返回 String 以防未命中。您可以轻松修复您的过滤器:

val pairs = lines.map(
  l => (if (l.split(",")(1).toInt < 60) {"rest"} else if (l.split(",")(1).toInt > 110) {"sport"}, 10))
    .filter(_._1 != ())

() 在 scala 中是 Unit 类型的标识。

但这不是正确的方法，真的。您仍然会得到 (Unit, Int) 的元组作为结果。您正在使用此 if 语句丢失类型。

正确的方法是先过滤您的数据，然后进行详尽的if:

val pairs =
  lines.map(_.split(",")(1).toInt)
    .filter(hr => hr < 60 || hr > 110)
    .map(hr => (if (hr < 60) "rest" else "sport", 10))

或者使用collect，在spark is the shortcut for .filter.map:

val pairs =
  lines.map(_.split(",")(1).toInt)
    .collect{
      case hr if hr < 60 => "rest" -> 10
      case hr if hr > 110 => "sport" -> 10
    }

可能这个变体更具可读性。

另外，请注意我是如何将 split 移动到单独的步骤中的。这样做是为了避免在第二个 if 分支中第二次调用 split。

更新。另一种方法是使用 flatMap，如评论中所建议的:

val pairs =
  lines.flatMap(_.split(",")(1).toInt match{
      case hr if hr < 60 => Some("rest" -> 10)
      case hr if hr > 110 => Some("sport" -> 10)
      case _ => None
    })

它可能会或可能不会更有效，因为它避免了 filter 步骤，但在 Option 中添加了包装和展开元素。您可以测试不同方法的性能并告诉我们结果。