DataFrame 與 RDD 的互動

Spark SQL它支援兩種不同的方式轉換已經存在的RDD到DataFrame

方法一

第一種方式是使用反射的方式，用反射去推倒出來RDD裡面的schema。這個方式簡單，但是不建議使用，因為在工作當中，使用這種方式是有限制的。
對於以前的版本來說，case class最多支援22個欄位如果超過了22個欄位，我們就必須要自己開發一個類，實現product接口才行。因此這種方式雖然簡單，但是不通用；因為生產中的欄位是非常非常多的，是不可能只有20來個欄位的。
示例：

/**
  * convert rdd to dataframe 1
  * @param
 
 spark
  */
private def runInferSchemaExample(spark:SparkSession): Unit ={
  import spark.implicits._
  val rdd = spark.sparkContext.textFile("E:/大資料/data/people.txt")
  val df = rdd.map(_.split(","))
              .map(x => People(x(0), x(1).trim.toInt))  //將rdd的每一行都轉換成了一個people
              .toDF         //必須先匯入import spark.implicits._  不然這個方法會報錯
 

  df.show()

  df.createOrReplaceTempView("people")

  // 這個DF包含了兩個欄位name和age
  val teenagersDF = spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")
  // teenager(0)代表第一個欄位
  // 取值的第一種方式：index from zero
  teenagersDF.map(teenager => "Name: " + teenager(0)).show()
  // 取值的第二種方式：byName
  teenagersDF.map(teenager => "Name: "
 
 + teenager.getAs[String]("name") + "," + teenager.getAs[Int]("age")).show()
}

// 注意：case class必須定義在main方法之外；否則會報錯
case class People(name:String, age:Int)

方法二

建立一個DataFrame，使用程式設計的方式 這個方式用的非常多。通過程式設計方式指定schema ，對於第一種方式的schema其實定義在了case class裡面了。
官網解讀：
當我們的case class不能提前定義(因為業務處理的過程當中，你的欄位可能是在變化的),因此使用case class很難去提前定義。
使用該方式建立DF的三大步驟：

1. Create an RDD of Rows from the original RDD;
2. Create the schema represented by a StructType matching the structure of Rows in the RDD created in Step 1.
3. Apply the schema to the RDD of Rows via createDataFrame method provided by SparkSession.

示例：

/**
  * convert rdd to dataframe 2
  * @param spark
  */
private def runProgrammaticSchemaExample(spark:SparkSession): Unit ={
  // 1.轉成RDD
  val rdd = spark.sparkContext.textFile("E:/大資料/data/people.txt")

  // 2.定義schema，帶有StructType的
  // 定義schema資訊
  val schemaString = "name age"
  // 對schema資訊按空格進行分割
  // 最終fileds裡包含了2個StructField
  val fields = schemaString.split(" ")
                            // 欄位型別，欄位名稱判斷是不是為空
                           .map(fieldName => StructField(fieldName, StringType, nullable = true))
  val schema = StructType(fields)

  // 3.把我們的schema資訊作用到RDD上
  //   這個RDD裡面包含了一些行
  // 形成Row型別的RDD
  val rowRDD = rdd.map(_.split(","))
                  .map(x => Row(x(0), x(1).trim))
  // 通過SparkSession建立一個DataFrame
  // 傳進來一個rowRDD和schema，將schema作用到rowRDD上
  val peopleDF = spark.createDataFrame(rowRDD, schema)

  peopleDF.show()
}

【擴充套件】生產上建立DataFrame的程式碼舉例

在實際生產環境中，我們其實選擇的是方式二這種進行建立DataFrame的，這裡將展示部分程式碼：

Schema的定義

object AccessConvertUtil {

  val struct = StructType(
    Array(
      StructField("url",StringType),
      StructField("cmsType",StringType),
      StructField("cmsId",LongType),
      StructField("traffic",LongType),
      StructField("ip",StringType),
      StructField("city",StringType),
      StructField("time",StringType),
      StructField("day",StringType)
    )
  )

  /**
    * 根據輸入的每一行資訊轉換成輸出的樣式
    */
  def parseLog(log:String) = {

    try {
      val splits = log.split("\t")

      val url = splits(1)
      val traffic = splits(2).toLong
      val ip = splits(3)

      val domain = "http://www.imooc.com/"
      val cms = url.substring(url.indexOf(domain) + domain.length)
      val cmsTypeId = cms.split("/")

      var cmsType = ""
      var cmsId = 0l
      if (cmsTypeId.length > 1) {
        cmsType = cmsTypeId(0)
        cmsId = cmsTypeId(1).toLong
      }

      val city = IpUtils.getCity(ip)
      val time = splits(0)
      val day = time.substring(0,10).replace("-","")

      //這個Row裡面的欄位要和struct中的欄位對應上
      Row(url, cmsType, cmsId, traffic, ip, city, time, day)
    } catch {
      case e: Exception => Row(0)
    }

  }

}

建立DataFrame

object SparkStatCleanJob {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("SparkStatCleanJob")
      .master("local[2]").getOrCreate()

    val accessRDD = spark.sparkContext.textFile("/Users/lemon/project/data/access.log")

    //accessRDD.take(10).foreach(println)

    //RDD ==> DF，建立生成DataFrame
    val accessDF = spark.createDataFrame(accessRDD.map(x => AccessConvertUtil.parseLog(x)),
      AccessConvertUtil.struct)

    accessDF.coalesce(1).write.format("parquet").mode(SaveMode.Overwrite)
            .partitionBy("day").save("/Users/lemon/project/clean")

    spark.stop()
  }

}