Блог

Эта функция кажется допустимой для моей IDE:

 def zip[T, U](rdd1:RDD[T], rdd2:RDD[U]) : RDD[(T,U)] = {
    rdd1
      .zipWithIndex
      .map(_.swap)
      .join(
        rdd2
          .zipWithIndex
          .map(_.swap))
      .values
}
  

Но когда я компилирую, я получаю :

соединение значений не является членом org.apache.spark.rdd.RDD[(Long, T)] возможная причина: может быть, перед «объединением значений» отсутствует точка с запятой? .присоединиться(

Я нахожусь в Spark 1.6, я уже пытался импортировать org.apache.spark.rdd.RDD._ и код внутри функции работает хорошо, когда он напрямую используется на двух RDD вне определения функции.

Есть идеи?

Если вы измените подпись:

 def zip[T, U](rdd1:RDD[T], rdd2:RDD[U]) : RDD[(T,U)] = {
  

в:

 def zip[T : ClassTag, U: ClassTag](rdd1:RDD[T], rdd2:RDD[U]) : RDD[(T,U)] = {
  

Это будет скомпилировано.

Почему? join это метод PairRDDFunctions (ваш RDD неявно преобразуется в этот класс), который имеет следующую подпись:

 class PairRDDFunctions[K, V](self: RDD[(K, V)])
  (implicit kt: ClassTag[K], vt: ClassTag[V], ord: Ordering[K] = null)
  

Это означает, что его конструктор ожидает неявные значения типов ClassTag[T] и ClassTag[U] , поскольку они будут использоваться в качестве типов значений ( V в PairRDDFunctions определении). Ваш метод не знает, что T такое и U , и поэтому не может предоставить соответствующие неявные значения. Это означает, что неявное преобразование в PairRDDFunctions «сбой» (компилятор не выполняет преобразование) и, следовательно, метод join не может быть найден.

Добавление [K : ClassTag] — это сокращение для добавления неявного аргумента implicit kt: ClassTag[K] к методу, который затем используется компилятором и передается конструктору PairRDDFunctions .

Подробнее о классовых тегах и о том, для чего они нужны, см. В Этой хорошей статье .