如何使用用户定义的类和toDF将RDD转换为数据帧

Question

当我尝试通过sbt pack创建以下包时：

import org.apache.spark.sql.SparkSession

class Log(val cip: String, val scstatus: Int) {
    var src: String = cip
    var status: Int = scstatus
}

object IISHttpLogs {
  def main(args: Array[String]) {
    val logFiles = "D:/temp/tests/wwwlogs" 
    val spark = SparkSession.builder.appName("LogParser").getOrCreate()
    val sc = spark.sparkContext;
    sc.setLogLevel("ERROR")

    val logs = sc.textFile(logFiles)        

    import spark.implicits._
    val rowDF = logs.filter(l => !l.startsWith("#"))
        .map(l => l.split(" "))
        .map(c => new Log(c(8), c(11).trim.toInt))
        .toDF();
    println(s"line count: ${rowDF.count()}")        
    rowDF.createOrReplaceTempView("rows")
    val maxHit = spark.sql("SELECT top 1 src, count(*) FROM rows group by src order by count(*) desc")
    maxHit.show()

    spark.stop()
  }
}

我得到以下错误：

值toDF不是组织的成员。阿帕奇。火花rdd。RDD[日志]

我尝试了以下几种方法：

• 托德弗洛格

我就是不能编译我的代码。

欢迎任何线索来克服这个错误。

我很好地阅读了从case类生成Spark StructType/Schema，并编写了：

val schema =
    StructType(
        StructField("src", StringType, false) ::
        StructField("status", IntegerType, true) :: Nil)

val rowRDD = logs.filter(l => !l.startsWith("#"))
    .map(l => l.split(" "))
    .map(c => Row(c(8), c(11).trim.toInt));

val rowDF = spark.sqlContext.createDataFrame(rowRDD, schema); 

但这样做时，我不使用Log类。我想知道是否有一种方法可以通过使用定义的日志类获取数据帧，或者官方/最佳方法是使用行类？

例如，我不会写：

val rowRDD = logs.filter(l => !l.startsWith("#"))
    .map(l => l.split(" "))
    .map(c => new Log(c(8), c(11).trim.toInt));
val rowDF = spark.sqlContext.createDataFrame(
    rowRDD,
    ScalaReflection.schemaFor[Log].dataType.asInstanceOf[StructType]);

我就是不明白为什么？

Answer 1

你必须使用案例类。至少对我有用：

case class Log(cip: String,  scstatus: Int)
//...
.map(c =>  Log(c(8), c(11).trim.toInt) // ommit 'new'
.toDF()

我不太确定这是否是一般规则。但是在Dataset API的公告中，明确提到了case类的使用：

Spark 1.6支持为各种类型自动生成编码器，包括基本类型（例如字符串、整数、长）、Scala case类和Java bean。(https://databricks.com/blog/2016/01/04/introducing-apache-spark-datasets.html)

如果不能使用case类，这个答案似乎是合适的。