Spark Scala 通用 min() 函数答案

【问题标题】：Spark Scala generic min() functionSpark Scala 通用 min() 函数
【发布时间】：2015-06-03 16:57:41
【问题描述】：

如何在 Spark 中创建一个泛型 min() 函数，它返回与所使用的泛型相同类型的值？

这是我对双打和弦乐的看法：

def minDouble(rdd: RDD[Map[String, String]], field: String): Double = {
  rdd.map(row => row(field).toDouble).min()
}

def minString(rdd: RDD[Map[String, String]], field: String): String = {
  rdd.map(row => row(field)).min()
}

如何使用泛型使其成为单个函数？并使其可扩展，以便我可以添加其他类型。

这是我的尝试：

def minGeneric[V : Manifest](rdd: RDD[Map[String, String]], field: String)(implicit ord: Ordering[V]): V = {
  rdd.map(row => cast[V](row(field))).min().get
}

它使用来自Writing a generic cast function Scala 的cast() 函数

当我运行这个时，我得到一个NoSuchElementException: None.get：

Exception in thread "main" java.util.NoSuchElementException: None.get
        at scala.None$.get(Option.scala:313)
        at scala.None$.get(Option.scala:311)
        at SimpleApp$.statMinGeneric(SimpleApp.scala:67)
        at SimpleApp$.main(SimpleApp.scala:34)
        at SimpleApp.main(SimpleApp.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:483)
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:569)
        at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:166)
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:189)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:110)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

为什么这里返回 None ？我采取了正确的方法吗？提前致谢。

【问题讨论】：

你有没有像_.toDouble这样的传递函数的选项？
@DmitryGinzburg 你能举个例子说明它是如何工作的吗？
也许它不匹配或者它没有说它是那种类型。您能否发布您的示例用法？
@JustinPiphony 我是这样使用它的：minGeneric[Double](maps, "wage")
我不知道如果没有像 Dmitry 建议的那样进行投射的提示，这是可能的。字符串不能直接转换为另一种类型。

标签： scala generics apache-spark

【解决方案1】：

如果您可以选择将其他参数传递给您的函数，您可以执行以下操作：

def minCasted[A](caster: String => A)(rdd: RDD[Map[String, String]], field: String): A = {
    caster(rdd.map(_(field)).min())
}

def minDouble = minCasted(_.toDouble) _

def minString = minCasted(identity[String]) _

【讨论】：