dataframe是在spark1.3.0中推出的新的api，這讓spark具備了處理大規模結構化資料的能力，在比原有的RDD轉化方式易用的前提下，據說計算效能更還快了兩倍。spark在離線批處理或者實時計算中都可以將rdd轉成dataframe進而通過簡單的sql命令對資料進行操作，對於熟悉sql的人來說在轉換和過濾過程很方便，甚至可以有更高層次的應用，比如在實時這一塊，傳入kafka的topic名稱和sql語句，後臺讀取自己配置好的內容欄位反射成一個class並利用出入的sql對實時資料進行計算，這種情況下不會spark streaming的人也都可以方便的享受到實時計算帶來的好處
　　　
　　　下面的示例為讀取本地檔案成rdd並隱式轉換成dataframe對資料進行查詢，最後以追加的形式寫入mysql表的過程,scala程式碼示例如下

import java.sql.Timestamp

import org.apache.spark.sql.{SaveMode, SQLContext}
import org.apache.spark.{SparkContext, SparkConf}

object DataFrameSql {
  case class memberbase(data_date:Long,memberid:String,createtime:Timestamp,sp:Int)extends Serializable{
    override  def toString: String="%d\t%s\t%s\t%d"
 
.format(data_date,memberid,createtime,sp)
  }

  def main(args:Array[String]): Unit ={
    val conf = new SparkConf()
    conf.setMaster("local[2]")
// ----------------------
    //引數 spark.sql.autoBroadcastJoinThreshold 設定某個表是否應該做broadcast，預設10M，設定為-1表示禁用
    //spark.sql.codegen 是否預編譯sql成java位元組碼，長時間或頻繁的sql有優化效果
 

    // spark.sql.inMemoryColumnarStorage.batchSize 一次處理的row數量，小心oom
    //spark.sql.inMemoryColumnarStorage.compressed 設定記憶體中的列儲存是否需要壓縮
// ----------------------
    conf.set("spark.sql.shuffle.partitions","20") //預設partition是200個
    conf.setAppName("dataframe test")

    val sc =  new SparkContext(conf)
    val sqc = new SQLContext(sc)

    val ac  = sc.accumulator(0,"fail nums")
    val file = sc.textFile("src\\main\\resources\\000000_0")
    val log = file.map(lines => lines.split("  ")).filter(line =>
      if (line.length != 4) { //做一個簡單的過濾
        ac.add(1)
        false
      } else true)
      .map(line => memberbase(line(0).toLong, line(1),Timestamp.valueOf(line(2)), line(3).toInt))
    // 方法一、利用隱式轉換
    import sqc.implicits._
    val dftemp = log.toDF()  // 轉換
    /*
     方法二、利用createDataFrame方法,內部利用反射獲取欄位及其型別
     val dftemp =  sqc.createDataFrame(log)
     */
    val df = dftemp.registerTempTable("memberbaseinfo")
    /*val sqlcommand ="select date_format(createtime,'yyyy-MM')as mm,count(1) as nums " +
      "from memberbaseinfo group by date_format(createtime,'yyyy-MM') " +
      "order by nums desc,mm asc "*/

    val sqlcommand="select * from memberbaseinfo"
    val sel = sqc.sql(sqlcommand)
    val prop = new java.util.Properties
    prop.setProperty("user","etl")
    prop.setProperty("password","xxx")

    // 呼叫DataFrameWriter將資料寫入mysql
    val dataResult = sqc.sql(sqlcommand).write.mode(SaveMode.Append).jdbc("jdbc:mysql://localhost:3306/test","t_spark_dataframe_test",prop) // 表可以不存在
    println(ac.name.get+"  "+ac.value)
    sc.stop()
  }
}

上面程式碼textFile中的示例資料如下，資料來自hive，欄位資訊分別為 分割槽號、使用者id、註冊時間、第三方號

20160309  45386477  2012-06-12 20:13:15  901438
20160309  45390977  2012-06-12 22:38:06  901036
20160309  45446677  2012-06-14 21:57:39  901438
20160309  45464977  2012-06-15 13:42:55  901438
20160309  45572377  2012-06-18 14:55:03  902606
20160309  45620577  2012-06-20 00:21:09  902606
20160309  45628377  2012-06-20 10:48:05  901181
20160309  45628877  2012-06-20 11:10:15  902606
20160309  45667777  2012-06-21 18:58:34  902524
20160309  45680177  2012-06-22 01:49:55  
20160309  45687077  2012-06-22 11:23:22  902607

這裡注意欄位型別對映，即case class類到dataframe對映，從官網的截圖如下