sql - 如何将 Spark SchemaRDD 转换为我的案例类的 RDD？

Question

在 Spark 文档中，很清楚如何从您自己的案例类的 RDD 创建 Parquet 文件； (来自文档)

val people: RDD[Person] = ??? // An RDD of case class objects, from the previous example.

// The RDD is implicitly converted to a SchemaRDD by createSchemaRDD, allowing it to be stored using Parquet.
people.saveAsParquetFile("people.parquet")

但不清楚如何转换回来，我们真的需要一个方法readParquetFile，我们可以在其中执行以下操作:

val people: RDD[Person] = sc.readParquestFile[Person](path)

定义案例类的那些值是由该方法读取的值。

Answer 1

一个简单的方法是提供您自己的转换器(Row) => CaseClass。这是更多的手册，但如果你知道你在读什么，它应该是相当简单的。

这是一个例子:

import org.apache.spark.sql.SchemaRDD

case class User(data: String, name: String, id: Long)

def sparkSqlToUser(r: Row): Option[User] = {
    r match {
      case Row(time: String, name: String, id: Long) => Some(User(time,name, id))
      case _ => None
    }
}

val parquetData: SchemaRDD = sqlContext.parquetFile("hdfs://localhost/user/data.parquet")

val caseClassRdd: org.apache.spark.rdd.RDD[User] = parquetData.flatMap(sparkSqlToUser)