Scala joinWithCassandraTable 结果（或 CassandraTableScanRDD）到数据集

Question

我正在使用 Datastax spark-cassandra-connector 访问 Cassandra 中的一些数据。

为了能够有效地访问查询所需的所有数据，我必须使用joinWithCassandraTable 方法从一堆分区中获取数据。这给了我一个 com.datastax.spark.connector.rdd.CassandraTableScanRDD 类的对象（或类似的，为了测试我实际上只是使用标准的 sc.cassandraTable(ks, tbl) 方法来读取数据）。

问题是，我需要在结果对象上使用的所有方法都需要一个 org.apache.spark.sql.Dataset 类的对象。

我已经做了很多搜索，但找不到任何帮助 - 我找到的最接近的是 this 类似问题，我认为它没有得到充分回答，因为它忽略了访问所有必要数据的推荐方法是使用joinWithCassandraTable。

我也是 java 和 scala 的新手，如果我有点慢，很抱歉。任何帮助都将不胜感激，因为我在这一点上非常困难。

谢谢， 阿基尔

Answer 1

您可以做的是将您的 RDD 读入 RDD[Row]，然后将其更改为 DataFrame。我们唯一的问题是我们还需要 Schema。所以让我们分两步来做。

首先让我们以编程方式从我们的连接目标中获取架构

val schema = spark.read.cassandraFormat("dogabase", "test").load.schema

/**
schema: org.apache.spark.sql.types.StructType = 
StructType(StructField(owner,StringType,true), 
StructField(dog_id,IntegerType,true), 
StructField(dog_age,IntegerType,true), 
StructField(dog_name,StringType,true))
**/

然后我们可以从我们的 Cassandra 驱动程序中创建 org.apache.spark.sql.Row 对象 行。

import org.apache.spark.sql.Row
val joinResult = 
  sc.parallelize(Seq(Tuple1("Russ")))
    .joinWithCassandraTable("test", "dogabase")
    .map{ case(_, cassandraRow) => Row(cassandraRow.columnValues:_*)} //Unpack our Cassandra row values into a spark.sql.Row

现在我们有了一个模式和一个 RDD[Row]，我们可以使用 spark session 的 createDataFrame 方法

val dataset = spark.createDataFrame(joinResult, schema)
dataset.show

/**
+-----+------+-------+--------+
|owner|dog_id|dog_age|dog_name|
+-----+------+-------+--------+
| Russ|     1|     10|    cara|
| Russ|     2|     11|sundance|
+-----+------+-------+--------+
**/

如果你不相信 DataFrame 是一个数据集

dataset.getClass
Class[_ <: org.apache.spark.sql.DataFrame] = class org.apache.spark.sql.Dataset

编辑：可能需要的转换器

某些 Cassandra 类型不是 Spark Rows 的有效基础，因此您可能需要转换它们。这可以通过编写一个快速转换函数来完成。不幸的是，SCC 使用的内置转换是一种内部表示，因此我们无法使用这些转换。

def convertToSpark(element:Any): Any = {
  case time: org.joda.time.LocalDate => time.toDateTimeAtStartOfDay().toDate //Convert to java.util.Date
  case other => other
}

然后在制作行时

cassandraRow.columnValues.map(convertToSpark):_*