scala将CassandRatable结果连接到dataframe

Question

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

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

正如我所看到的，您在联接的左侧使用了dataframe。与其使用使用RDD API的JoinWithCassandRatable，我建议使用支持Dataframe API中join的Spark Cassandra连接器2.5.x(2.5.1是最新版本），并直接使用它。这非常简单，您只需使用--conf spark.sql.extensions=com.datastax.spark.connector.cassandrasParkExtensions启动您的作业就可以激活此功能，之后，代码只需在Dataframes上使用普通联接：

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")

这里有自述文件的完整源代码。