将 RDD 上传到现有 Cassandra 时出错 table

Question

我使用 Dataframe 连接器函数从 Cassandra table 中提取内容。之后，我对数据帧执行了一些过滤（无转换），然后我想将其再次写入现有的 table。为此，我需要先将 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]))

我已经检查过，所有元素的类型都已正确设置。现在我想使用 results.saveToCassandra("labels", "results", SomeColumns(...)) 将此 RDD 上传到 Cassandra，但出现以下错误。

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.apply(<console>:65)
    at $$$$b42ed3d02f91ffa45dcf288dd693450$$$$$anonfun.apply(<console>:65)
    at scala.collection.Iterator$$anon.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？

编辑：

我根据第一个建议更新了代码：

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.

Answer 1

asInstance 不会将您的数据变成它不是的东西。相反，您应该转换它（并且通常使用类型化的 getter 而不是 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 类型。

将 RDD 上传到现有 Cassandra 时出错 table

Error uploading RDD to existing Cassandra table

apache-spark

cassandra-3.0

spark-cassandra-connector