scala joinWithCassandraTable 结果到数据框

Question

我正在使用 Datastax spark-Cassandra-connector 访问 Cassandra 中的一些数据。 我的要求是将 RDD 与 Cassandra 表连接，获取结果并将其存储在 hive 表中。

我正在使用 joinWithCassandraTable 加入 cassadra 表。加入后，产生的 RDD 如下所示

com.datastax.spark.connector.rdd.CassandraJoinRDD[org.apache.spark.sql.Row, 
com.datastax.spark.connector.CassandraRow] = 
CassandraJoinRDD[17] at RDD at CassandraRDD.scala:19

我尝试了以下步骤来转换为数据框，但没有一种方法有效。

val data=joinWithRDD.map{
   case(_, cassandraRow) =>    Row(cassandraRow.columnValues:_*)
}

sqlContext.createDataFrame(data,schema)

我遇到了错误

java.lang.ClassCastException: cannot assign instance of
   scala.collection.immutable.List$SerializationProxy to field 
   org.apache.spark.rdd.RDD.org$apache$spark$rdd$RDD$$dependencies_ of 
   type scala.collection.Seq in instance of org.apache.spark.rdd.MapPartitionsRDD

您能帮我将joinWithCassandraTable 转换为数据框吗？

Answer 1

如我所见，您在联接的左侧使用数据框。我建议不要使用使用 RDD API 的 joinWithCassandraTable，而是使用 has support for join in the Dataframe API 的 Spark Cassandra 连接器 2.5.x（2.5.1 是最新的），并直接使用它。真的很简单，你只需要用--conf spark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions开始你的工作来激活这个functionality，之后，代码就只是在数据帧上使用普通的连接：

val parsed = ...some dataframe...
val cassandra = spark.read
  .format("org.apache.spark.sql.cassandra")
  .options(Map("table" -> "stock_info", "keyspace" -> "test"))
  .load

// we can use left join to detect what data is incorrect - if we don't have some data in the
// Cassandra, then symbol field will be null, so we can detect such entries, and do something with that
// we can omit the joinType parameter, in that case, we'll process only data that are in the Cassandra
val joined = parsed.join(cassandra, cassandra("symbol") === parsed("ticker"), "left")
   .drop("ticker")

带有自述文件的完整源代码是here。