package com.lyzx.day19

import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.Row

class T1 {
  def f1(sc:SparkContext): Unit ={
    val sqlCtx = new SQLContext(sc)
    import  sqlCtx.implicits._

    val myDf = sc.textFile("./Person.txt")
                 .map(x=>x.split(","))
                 .map(x=>My(x(0),x(1),x(2).toInt)).toDF()

    myDf.registerTempTable("myUser")

    val result = sqlCtx.sql("select * from myUser where id >= 2")
    result
      //這兒的_(x)指的是當前一條記錄的第(x+1)個屬性
      //也就是說要取某一條記錄的某個屬性可以通過下標的形式(_(index))也可以通過列名的形式(_.getAs[T](colNameStr))
//        .map(_(0))
      .map(_.getAs[String]("name"))
      .foreach(println)
  }


  def f2(sc:SparkContext): Unit ={
    val sqlCtx = new SQLContext(sc)
    val rdd= sc.textFile("./person.txt")
              .map(_.split(","))
              .map(x=>Row(x(0),x(1),x(2)))

    //這兒有侷限因為傳入的是一個數組，即每一個元素都一樣，如果是每個列的
    val schema = StructType(Array(StructField("id",StringType,true),StructField("name",StringType,true),StructField("age",StringType,true)))
    val df = sqlCtx.createDataFrame(rdd,schema)
    df.registerTempTable("person")

    sqlCtx.sql("select name from person where id >=2")
          .foreach(println)

  }

  /**
   * 總結對於兩種RDD轉DataFrame的方式
   * 結果都是一樣，唯一的不同就是定義schema的方式，
   *  反射的方式  Schema是寫死的即實體類
   *    優點:使用簡單
   *    缺點:如果要修改schema就要是修改實體類即修改麻煩
   *  動態的方式
   *    優點:靈活一些,不過對於列的型別也不夠靈活，可以動態的指定列的名稱
   *    缺點:使用時麻煩一些
   */
}

//這個類要放在外面而且前面還要加case
//這是因為將一行資料資料轉換為My物件後需要序列化
case class My(id:String,name:String,age:Int)
object T1{
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("day18").setMaster("local")
    val sc = new SparkContext(conf)

    val t = new T1
    t.f1(sc)
    sc.stop()
  }
}