scala - 如何使用 toDF() 将自定义 Java 类对象的 RDD 转换为 DataFrame？

Question

我正在尝试使用 toDF() 将 Spark RDD 转换为 Spark SQL 数据帧.我多次成功使用此函数，但在这种情况下，我收到编译器错误:

error: value toDF is not a member of org.apache.spark.rdd.RDD[com.example.protobuf.SensorData]

这是我的代码如下:

// SensorData is an auto-generated class
import com.example.protobuf.SensorData
def loadSensorDataToRdd : RDD[SensorData] = ???

object MyApplication {
  def main(argv: Array[String]): Unit = {

    val conf = new SparkConf()
    conf.setAppName("My application")
    conf.set("io.compression.codecs", "com.hadoop.compression.lzo.LzopCodec")
    val sc = new SparkContext(conf)

    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    import sqlContext.implicits._

    val sensorDataRdd = loadSensorDataToRdd()
    val sensorDataDf = sensorDataRdd.toDF() // <-- CAUSES COMPILER ERROR
  }
}

我猜测问题出在 SensorData 类上，它是一个从 Protocol Buffer 自动生成的 Java 类。 .我该怎么做才能将 RDD 转换为数据帧？

Answer 1

编译错误的原因是没有Encoder在范围内转换 RDD与 com.example.protobuf.SensorData到 Dataset的 com.example.protobuf.SensorData .
Encoders ( ExpressionEncoders 准确)用于转换 InternalRow根据模式将对象转换为 JVM 对象(通常是 case 类或 Java bean)。

希望你能创建一个Encoder对于使用 org.apache.spark.sql.Encoders 的自定义 Java 类对象的 bean方法。

Creates an encoder for Java Bean of type T.

类似于以下内容:

import org.apache.spark.sql.Encoders
implicit val SensorDataEncoder = Encoders.bean(classOf[com.example.protobuf.SensorData])

如 SensorData使用不受支持的类型，您必须map RDD[SensorData]到 RDD一些更简单的类型，例如字段的元组，然后才期望 toDF工作。