Spark DataFrame zipWithIndex

Question

我正在使用 DataFrame 读取 .parquet 文件，但不是将它们变成 rdd 来进行我想要对它们进行的正常处理。

所以我有我的文件：

val dataSplit = sqlContext.parquetFile("input.parquet")
val convRDD = dataSplit.rdd 
val columnIndex = convRDD.flatMap(r => r.zipWithIndex)

即使我从数据帧转换为 RDD，也会出现以下错误：

:26: 错误：值 zipWithIndex 不是 org.apache.spark.sql.Row

任何人都知道如何做我想做的事情，本质上是尝试获取值和列索引。

我在想这样的事情：

val dataSplit = sqlContext.parquetFile(inputVal.toString)
val schema = dataSplit.schema
val columnIndex = dataSplit.flatMap(r => 0 until schema.length

但由于不确定如何对 zipWithIndex 执行相同的操作，因此卡在了最后一部分。

Answer 1

您可以简单地将Row 转换为Seq：

convRDD.flatMap(r => r.toSeq.zipWithIndex)

这里要注意的重要一点是提取类型信息变得很棘手。 Row.toSeq 返回 Seq[Any] 并且结果 RDD 是 RDD[(Any, Int)]。