0

我正在尝试通过 Spark 2.0 中的函数访问 HashMap，但如果我并行化列表，它将失败。如果我不这样做，它会起作用，如果我不使用案例类，它会起作用。

这是我正在尝试做的一些示例代码：

case class TestData(val s: String)

def testKey(testData: TestData) {
  println(f"Current Map: $myMap")
  println(f"Key sent into function: $testData")
  println("Key isn't found in Map:")
  println(myMap(testData)) // fails here
}

val myList = sc.parallelize(List(TestData("foo")))
val myMap = Map(TestData("foo") -> "bar")
myList.collect.foreach(testKey) // collect to see println

这是确切的输出：

Current Map: Map(TestData(foo) -> bar)
Key sent into function: TestData(foo)
Key isn't found in Map:
java.util.NoSuchElementException: key not found: TestData(foo)

上面的代码与我正在尝试做的类似，除了案例类更复杂并且 HashMap 将列表作为值。同样在上面的示例中，我使用“收集”以便输出打印语句。样品在没有收集的情况下仍然给出相同的错误，但没有打印。

hashCodes 已经匹配，但我尝试为案例类覆盖 equals 和 hashCode，同样的问题。

这是使用 Databricks，所以我不相信我可以访问 REPL 或 spark-submit。

4

2 回答 2

1

感谢指出类似问题的评论，该问题涉及 Spark 问题，这使我为我的案例找到了这个解决方案：

case class TestData(val s: String) {
  override def equals(obj: Any) = obj.isInstanceOf[TestData] && obj.asInstanceOf[TestData].s == this.s
}

覆盖等号以包含 isInstanceOf 解决了该问题。它可能不是最好的解决方案，但绝对是最简单的解决方法。

于 2017-01-11T21:42:32.190 回答
-1

你的逻辑是循环和错误的。您正在将相同的 RDD 传递给 Map 并使用 TestData 进行调用。更新它以使其顺序如下：

case class TestData(val s: String)

def testKey(testData: TestData) {
  val myMap = Map(testData -> "bar")
  println(f"Current Map: $myMap")
  println(f"Key sent into function: $testData")
  println("Key isn't found in Map:")
  println(myMap(testData)) // fails here
}

val myList = sc.parallelize(List(TestData("foo")))
myList.collect.foreach(testKey)

它的输出是：

Current Map: Map(TestData(foo) -> bar)
Key sent into function: TestData(foo)
Key isn't found in Map:
bar

我希望这是你所期待的......

于 2017-01-12T05:49:34.027 回答