Объект класса case Scala «ключ не найден» при использовании в качестве ключа HashMap в Spark

Question

Я пытаюсь получить доступ к HashMap с помощью функции в Spark 2.0, но если я распараллеливаю List, он терпит неудачу. Если я этого не делаю, он работает, и если я не использую класс Case, он работает.

Вот некоторые примеры кода, что я пытаюсь сделать:

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) 

Код выше похож на то, что я пытаюсь сделать, за исключением того, что класс case более сложный, а HashMap имеет списки как значения. Также в вышеприведенном примере я использую «collect», чтобы выводились инструкции печати. Образец по-прежнему дает ту же ошибку без сбора, но без отпечатков.

Совпадение hashCodes уже, но я попытался переопределить равные и hashCode для класса case, той же проблемой.

Это использование Databricks, поэтому я не верю, что у меня есть доступ к REPL или spark-submit.

Answer 1

Благодаря комментариям за указание the similar question, которые шли к вопросу Спарк, что привело меня к этому решению для моего случая:

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

Переопределение Равных включить isInstanceOf исправляет эту проблему. Это может быть не лучшее решение, но это, безусловно, самый простой способ.

Answer 2

Ваша логика циклическая & неправильная. Вы передаете тот же RDD на карту &, вызывающую с 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 

Я надеюсь, что это то, что вы ожидаете ...