如何在Spark 2.x（Scala）中使用flatMap展开Cassandra UDT数据？

我来帮你搞定这个需求，在Spark 2.x的Scala环境下，我们可以通过两种方式用flatMap来展开Cassandra里的UDT数组数据，下面是具体的步骤和代码示例：

前提准备 #

首先确保你已经配置好了Spark与Cassandra的连接（依赖spark-cassandra-connector，版本要和Spark 2.x匹配，比如Spark 2.4.x对应spark-cassandra-connector_2.11-2.4.3），并且已经创建好对应UDT的样例类（方便Spark解析UDT结构）：

// 定义与Cassandra UDT对应的Scala样例类
case class MyUDT(s1: String, s2: String)

方式一：利用Dataset API（更简洁类型安全） #

先从Cassandra读取数据到DataFrame，再转换成Dataset进行flatMap操作：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.cassandra._
import spark.implicits._

// 初始化SparkSession
val spark = SparkSession.builder()
  .appName("FlattenCassandraUDT")
  .config("spark.cassandra.connection.host", "你的Cassandra主机地址")
  .getOrCreate()

// 读取Cassandra表数据（替换成你的keyspace和表名）
val rawDF = spark.read
  .format("org.apache.spark.sql.cassandra")
  .options(Map("keyspace" -> "your_keyspace", "table" -> "your_table"))
  .load()

// 用flatMap展开UDT数组
val flattenedDF = rawDF.as[(Int, Seq[MyUDT])].flatMap { case (id, udtList) =>
  // 把每个UDT元素和id组合成新的记录
  udtList.map(udt => (id, udt.s1, udt.s2))
}.toDF("id", "s1", "s2")

// 查看展开后的结果
flattenedDF.show()

执行后你会得到展开后的结果：

+---+-----+-----+
| id|   s1|   s2|
+---+-----+-----+
|  1|str11|str12|
|  1|str13|str14|
|  2|str21|str22|
|  2|str23|str24|
|  3|str31|str32|
|  3|str33|str44|
+---+-----+-----+

方式二：直接操作DataFrame的Row（无需样例类） #

如果不想定义样例类，也可以直接对DataFrame的Row对象进行flatMap操作：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.cassandra._
import org.apache.spark.sql.Row

val spark = SparkSession.builder()
  .appName("FlattenCassandraUDT")
  .config("spark.cassandra.connection.host", "你的Cassandra主机地址")
  .getOrCreate()

val rawDF = spark.read
  .format("org.apache.spark.sql.cassandra")
  .options(Map("keyspace" -> "your_keyspace", "table" -> "your_table"))
  .load()

val flattenedDF = rawDF.flatMap { row =>
  // 从Row中提取id和UDT数组
  val id = row.getInt(row.fieldIndex("id"))
  val udtRows = row.getSeq[Row](row.fieldIndex("some_data"))
  
  // 遍历UDT数组，生成新的记录
  udtRows.map { udtRow =>
    (id, udtRow.getString(udtRow.fieldIndex("s1")), udtRow.getString(udtRow.fieldIndex("s2")))
  }
}.toDF("id", "s1", "s2")

flattenedDF.show()

关键说明 #

• 两种方式核心都是用flatMap把每个id对应的UDT数组拆分成多条独立记录
• 方式一利用Dataset的类型推导，代码更简洁且类型安全；方式二更灵活，不需要提前定义样例类
• 注意替换代码中的Cassandra主机地址、keyspace和表名为你实际的配置

内容的提问来源于stack exchange，提问作者underwood