Ошибка при загрузке RDD в существующую таблицу Cassandra - PullRequest
Новая
       51

Ошибка при загрузке RDD в существующую таблицу Cassandra

0 голосов
/ 06 мая 2018

Я извлекаю содержимое из таблицы Cassandra, используя функции коннектора Dataframe. После этого я выполняю некоторую фильтрацию (без преобразований) на фрейме данных, а затем хочу снова записать его в существующую таблицу. Для этого мне нужно сначала превратить DF в RDD. Я делаю это с помощью следующей команды:

Для этого я использовал следующий код: 

results.rdd.map(row=> (row.get(0).asInstanceOf[String], row.get(1).asInstanceOf[String], row.get(2).asInstanceOf[String], row.get(3).asInstanceOf[java.util.UUID], row.get(4).asInstanceOf[String], row.get(5).asInstanceOf[String], row.get(6).asInstanceOf[Long], row.get(7).asInstanceOf[Set[String]], row.get(8).asInstanceOf[Array[Byte]], row.get(9).asInstanceOf[String], row.get(10).asInstanceOf[String], row.get(11).asInstanceOf[Set[String]], row.get(12).asInstanceOf[Set[String]], row.get(13).asInstanceOf[Set[String]], row.get(14).asInstanceOf[String], row.get(15).asInstanceOf[List[String]],row.get(16).asInstanceOf[String],row.get(17).asInstanceOf[String]))

Я проверил, и все элементы имеют соответствующий тип. Теперь я хочу загрузить этот RDD на Cassandra, используя results.saveToCassandra("labels", "results", SomeColumns(...)), но я получаю следующую ошибку. 

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 73.0 failed 4 times, most recent failure: Lost task 0.3 in stage 73.0 (TID 70606, hdp-worker-5.cloud.mwn.de, executor 172): java.lang.ClassCastException: scala.collection.mutable.WrappedArray$ofRef cannot be cast to scala.collection.immutable.Set
    at $$$$b42ed3d02f91ffa45dcf288dd693450$$$$$anonfun$1.apply(<console>:65)
    at $$$$b42ed3d02f91ffa45dcf288dd693450$$$$$anonfun$1.apply(<console>:65)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
    at com.datastax.spark.connector.util.CountingIterator.next(CountingIterator.scala:16)
    at com.datastax.spark.connector.writer.GroupingBatchBuilder.next(GroupingBatchBuilder.scala:106)
    at com.datastax.spark.connector.writer.GroupingBatchBuilder.next(GroupingBatchBuilder.scala:31)
    at scala.collection.Iterator$class.foreach(Iterator.scala:893)

Дело в том, что тип Cassandra для нескольких столбцов - это Set, что означает, что мой тип scala должен быть типом Set [String] (или TreeSet или HashSet - я тестировал их все), и я получаю эту ошибку. Как правильно отформатировать типы данных для загрузки на Cassandra?

Edit:

Я обновил код в соответствии с первым предложением:

results.rdd.map(row=> (row.getString(0), row.getString(1), row.getString(2), row.get[java.util.UUID](3), row.getString(4), row.getString(5), row.get[java.util.Date](6), row.get[Seq[String]](7).toSet, row.get[Array[Byte]](8), row.getString(9), row.getString(10), row.getSeq[String](11).toSet, row.get[Seq[String]](12).toSet, row.get[Seq[String]](13).toSet, row.getString(14), row.get[List[String]](15),row.getString(16),row.getString(17)))

Но я продолжаю получать ошибки, связанные с несколькими записями:

<console>:65: error: method get: (i: Int)Any does not take type parameters.

1 Ответ

0 голосов
/ 06 мая 2018

asInstance не превратит ваши данные в то, чем они не являются. Вместо этого вы должны конвертировать его (и обычно использовать типизированные геттеры, а не asInstanceOf): 

results.rdd.map(row => (
    row.getString(0),
    ...
    row.getSeq[String](7).toSet,
    ...,
    row.getSeq[String](11).toSet,
    ...
))

В этом коде могут быть другие ошибки, особенно: 

row.get(3).asInstanceOf[java.util.UUID]

выглядит странно, поскольку у Spark нет типа UUID.

Связано Искра, извлекающая значения из строки

...