какие методы искрового соединения зависят от завершения соединения

Question

Я хочу написать класс Key класса JavaPairRDD и реализовать свою логику соединения. Поэтому мне нужно переписать методы equals() и hashCode() для ключевого класса, или переписывать других. Например: Вы знаете, что JavaPairRDD.join (JavaPairRDD) очень прост, как раз при сопоставлении StringKey1.equals (StringKey2). Но для меня я надеюсь, что даже если подстрока key1 просто key2 также будет сопоставлена. Поэтому я думаю, что мне нужно написать класс переноса, чтобы рассказать Спарк, если они соответствуют моей логике, вы должны рассматривать их как один и тот же ключ и присоединяться к ним. например, «ab» следует присоединить к «abc».

Answer 1

Вам необходимо переопределить как equals, так и hashCode. ОДНАКО, вам нужно будет попытаться явно указать ваш объект other, так как совпадение шаблонов не работает. Я все еще пытаюсь понять, почему совпадение терпит неудачу только по типу (я предполагаю сериализацию). Также имейте в виду, что если ключевые объекты действительно не совпадают, тогда один будет переопределять другой в конечном выходе ... который может быть не таким, каким вы хотите.

case class Foo(x: String) { 
    override def equals(other: Any) = { 
    Option(other).map(_.asInstanceOf[Foo]).getOrElse(null) 
    match 
    { 
     case f : Foo => 
     x contains(f.x) 
     case _ => false 
    } 
    } 

    override def hashCode() = 1 
} 

sc.parallelize(List((Foo("abcdefg"),1))).join(sc.parallelize(List((Foo("abcdef"),1)))).collect 

Answer 2

Взлом семантику equals и hashCode, чтобы заставить искру присоединиться на вещи, которые не равны, является плохой идеей.

Spark полагается на ключ hashCode ключа для разделения данных на кластере. Принуждение константы hashCode заставит данные перейти к одному разделу, исключая возможность распределения вычислений в кластере. Это означает, что вычисление может быть/должно выполняться в одном узле, возможно, с использованием правильных одноузловых методов.

Эта проблема может быть решена с помощью Спарк, принимая другой подход:

val data1 = List("abcd" -> 4, "defghy" -> 6 , "wxxyyy" -> 3, "az" -> 24) 
val data2 = List("ab" -> 2, "fghz" -> 3 , "y" -> 1, "a" -> 2) 

val base = sparkContext.parallelize(data1) 
val jn = sparkContext.parallelize(data2) 

val cartesian = base.cartesian(jn) 

val joined = cartesian.collect{case ((k1,v1),(k2,v2)) if (k1.contains(k2)) => (k1,(v1,v2))} 

joined.collect 

Это приводит к:

res6: Array[(String, (Int, Int))] = Array((abcd,(4,2)), (defghy,(6,1)), (wxxyyy,(3,1)), (az,(24,2)))