spark 有三大引擎，spark core、sparkSQL、sparkStreaming，
spark core 的关键抽象是 SparkContext、RDD；
SparkSQL 的关键抽象是 SparkSession、DataFrame；
sparkStreaming 的关键抽象是 StreamingContext、DStream

@1 DataFrame/Dataset 转 RDD：val rdd1=testDF.rdd, val rdd2=testDS.rdd;
@2 RDD 转 DataFrame：// 一般用元组把一行的数据写在一起，然后在toDF中指定字段名

import spark.implicits._
val testDF = rdd.map {line=>
      (line._1,line._2)
    }.toDF("col1","col2");

@3 RDD 转 DataSet：// 核心就是要定义case class

import spark.implicits._  //导入Spark隐式转换，为RDD/Dataset/DF操作提供支持
case class Coltest(col1:String, col2:Int)  //定义样例类，作为Dataset的类型模板
val testDS = rdd.map{line=>Coltest(line._1,line._2)}.toDS;  //RDD映射为样例类对象，再转Dataset

spark.implicits._：必须导入，否则toDS()/toDF()方法无法使用，它为Scala集合和RDD提供了转换为Dataset/DataFrame的隐式方法。
case class Coltest：定义了Dataset中每条数据的结构（字段名+类型），这是Dataset强类型的核心——Dataset的每一条数据都是这个样例类的实例。
rdd.map{…}：将RDD中的元组（(String,Int)）映射为Coltest对象，把无类型的RDD转为有类型的结构。
toDS()：将映射后的RDD转换为Dataset[Coltest]，此时Dataset具备强类型特性，能在编译期检查类型错误。

@4 Dataset 转 DataFrame：// 这个转换简单，只是把 case class 封装成Row

import spark.implicits._
val testDF = testDS.toDF;  //Dataset直接转为DataFrame

toDF()：Dataset的内置方法，本质是将case class封装的强类型数据转为Row类型（DataFrame的核心是Dataset[Row]）。
转换后testDF的列名就是case class的字段名（col1、col2），列类型对应String、Int，无需额外指定。

@5 DataFrame 转 Dataset：// 每一列的类型后，使用as方法（as方法后面还是跟的case class，这个是核心），转成Dataset。这种方法就是在给出每一列的类型后，使用as方法，转成Dataset，这在数据类型是DataFrame又需要针对各个字段处理时极为方便

import spark.implicits._
case class Coltest(col1:String,col2:Int)extends Serializable  //继承Serializable保证序列化（分布式场景必需）
val testDS = testDF.as[Coltest];  //核心：as方法指定目标样例类类型

case class继承Serializable：Spark 分布式计算中，对象需要在节点间传输，序列化是必需的（虽然Scala的case class默认可序列化，但显式声明更规范）。
as[Coltest]：DataFrame转Dataset 的核心方法，作用是将无类型的Row数据映射回强类型的Coltest对象。
要求：DataFrame的列名、类型必须和case class完全匹配（列名大小写敏感，类型一致），否则会抛出类型转换异常。

@6 特别注意：在使用一些特殊操作时，一定要加上import spark.implicits._ 不然toDF、toDS无法使用

case class Person(name:String, age:Int, height:Int)
val arr = Array(("Jack", 28, 184), ("Tom", 10, 144), ("Andy", 16, 165))
val rdd1 = sc.makeRDD(arr)
val df = rdd1.toDF()
val df = rdd1.toDF("name", "age", "height")
df.as[Person];

完整可运行示例

import org.apache.spark.sql.SparkSession
// 定义样例类（建议继承Serializable，分布式场景必需）
case class Coltest(col1: String, col2: Int) extends Serializable
object DSDFConversion {
  def main(args: Array[String]): Unit = {
    // 1. 创建SparkSession（Spark 2.x+核心入口，sc是SparkContext的简写，从sparkSession获取）
    val spark = SparkSession.builder()
      .appName("DS-DF-Conversion")
      .master("local[*]")  // 本地运行模式
      .getOrCreate()

    import spark.implicits._  // 导入隐式转换，全局生效

    // 2. 定义样例类（建议放在main外，或加Serializable）
    case class Coltest(col1: String, col2: Int) extends Serializable

    // 3. 构造测试RDD
    val rdd = spark.sparkContext.parallelize(Seq(("a", 1), ("b", 2), ("c", 3)))

    // 4. RDD -> Dataset
    val testDS = rdd.map(line => Coltest(line._1, line._2)).toDS()
    println("=== Dataset内容 ===")
    testDS.show()  // 输出Dataset数据
    println(s"Dataset类型：${testDS.getClass.getName}")  // 打印类型

    // 5. Dataset -> DataFrame
    val testDF = testDS.toDF()
    println("\n=== DataFrame内容 ===")
    testDF.show()
    println(s"DataFrame类型：${testDF.getClass.getName}")  // 本质是Dataset[Row]

    // 6. DataFrame -> Dataset
    val testDS2 = testDF.as[Coltest]
    println("\n=== 转换回Dataset的内容 ===")
    testDS2.show()
    // 验证强类型：可以直接调用case class的字段
    testDS2.foreach(row => println(s"col1: ${row.col1}, col2: ${row.col2}"))

    // 关闭SparkSession
    spark.stop()
  }
}

SparkSession、SparkContext、SQLContext和HiveContext之间的区别。
1.SparkContext 是什么?
驱动程序使用SparkContext与集群进行连接和通信，它可以帮助执行Spark任务，并与资源管理器(如YARN 或Mesos)进行协调。
使用SparkContext，可以访问其他上下文，比如SQLContext和HiveContext。
使用SparkContext，我们可以为Spark作业设置配置参数。
如果您在spark-shell中，那么SparkContext已经为您提供了，并被分配给变量sc。
如果还没有SparkContext，可以先创建一个SparkConf。

//set up the spark configuration
val sparkConf = new SparkConf().setAppName("hirw").setMaster("yarn")
//get SparkContext using the SparkConf
val sc = new SparkContext(sparkConf)

2.SQLContext 是什么?
SQLContext是通往SparkSQL的入口。下面是如何使用SparkContext创建SQLContext。

// sc is an existing SparkContext.
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

一旦有了SQLContext，就可以开始处理DataFrame、DataSet等。

3.HiveContext 是什么?
HiveContext是通往hive入口。
HiveContext具有SQLContext的所有功能。
实际上，如果查看API文档，就会发现HiveContext扩展了SQLContext，这意味着它支持SQLContext支持的功能以及更多(Hive特定的功能)

public class HiveContext extends SQLContext implements Logging

下面是如何使用SparkContext获得HiveContext

// sc is an existing SparkContext.
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)

4.SparkSession 是什么?
SparkSession是在Spark 2.0中引入的，
它使开发人员可以轻松地使用它，这样我们就不用担心不同的上下文，
并简化了对不同上下文的访问。通过访问SparkSession，我们可以自动访问SparkContext。

下面是如何创建一个SparkSession

val spark = SparkSession
.builder()
.appName("hirw-test")
.config("spark.some.config.option", "some-value")
.master("local")
.getOrCreate()

SparkSession现在是Spark的新入口点，它替换了旧的SQLContext和HiveContext。注意，保留旧的SQLContext和HiveContext是为了向后兼容。

一旦我们访问了SparkSession，我们就可以开始使用DataFrame和Dataset了。

下面是我们如何使用Hive支持创建SparkSession。

val spark = SparkSession
.builder()
.appName("hirw-hive-test")
.config("spark.sql.warehouse.dir", warehouseLocation)
.enableHiveSupport()
.getOrCreate()

因此，如果您正在使用Spark 2.0或更高版本，建议使用SparkSession。

val spark = SparkSession.builder()
  .appName("Test DF")
  .config("config.option", "some-value")
  .master("local")
  .getOrCreate();

//import implicit DF,DS
import spark.implicits._ //这里的spark就可以正常使用

//csv文件
val spark = SparkSession.builder()
  .appName("csv reader")

//      .config("config.option", "some-value") 可以不要
      .master("local")
      .getOrCreate();
    //    header：首行是否作为列名
    //    sep：字段间的分隔符
    //    inferSchema：是否对字段类型进行推测。如果设置成False，默认读取后都按照文本字符处理。
    //val df1 = spark.read.csv("D:\\spark\\spark-core\\datas\\people1.csv");
    val df1 = spark
      .read
      .format("csv")
      .option("header","true")
      .option("multiLine","true")
      .option("encoding","gbk")  //utf-8
      .load("D:\\spark\\spark-core\\datas\\people1.csv")；

如下表对option里面的参数，进行介绍：

createDataFrame

import org.apache.spark.sql.{Row, SQLContext, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.types._

object createDataFrame {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("createDataFrame");
    val sc = new SparkContext(conf);
    val sqlContext = new org.apache.spark.sql.SQLContext(sc);
//    val sqlContext = new SQLContext(sc);

    val spark = SparkSession.builder()
      .appName("Test DF")
      .config("config.option", "some-value")
      .master("local")
      .getOrCreate()

    //import implicit DF,DS
    import spark.implicits._ //这里的spark就可以正常使用



    val schema = StructType(List(
      StructField("name", StringType, nullable = false),
      StructField("age", IntegerType, nullable = false),
      StructField("birthday", DateType, nullable = false)
    ));

    val rdd = sc.parallelize(Seq(
      Row("小明", 18, java.sql.Date.valueOf("1990-01-01")),
      Row("⼩芳", 20, java.sql.Date.valueOf("1999-02-01")) ))

    val df = sqlContext.createDataFrame(rdd, schema);
    df.show();
    println("*****************");
    val df1 = Seq(
      ("⼩明", 18, java.sql.Date.valueOf("1990-01-01")),
      ("⼩芳", 20, java.sql.Date.valueOf("1999-02-01"))
    ).toDF("name", "age", "birthday");
    df1.show()

    println("---------------");
    val rdd1 = spark.sparkContext.parallelize(List(1,2,3,4,5));
    val df2 = rdd1.map(x=>(x,x+2)).toDF("a","b");
    df2.show()
    sc.stop();
  }

}

spark sql

import org.apache.spark.sql.SparkSession

object spark_sql_transformation {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("Test DF")
      .config("config.option", "some-value")
      .master("local")
      .getOrCreate();


    //import implicit DF,DS
    import spark.implicits._ //这里的spark就可以正常使用
    var userDF = List((1, "张三", true, 18, 15000, 1))
      .toDF("id", "name", "sex", "age", "salary", "dept");
    userDF.createTempView("t_employee");
    val sql = "select * from t_employee where name = '张三'";
    spark.sql(sql).show();

    println("----------------");

    var userDF1= List( (1,"张三",true,18,15000,1),
      (2,"李四",false,18,12000,1),
      (3,"王五",false,18,16000,2))
      .toDF("id","name","sex","age","salary","dept");
    //构建视图
    userDF1.createTempView("t_employee1");
    val sql1=
      """
        |select dept ,avg(salary) as avg_slalary from t_employee1
        |group by dept order by avg_slalary desc """.stripMargin;
    spark.sql(sql1).show();

    println("******");

    // 开窗函数
    var df=List( (1,"zs",true,1,15000),
      (2,"ls",false,2,18000),
      (3,"ww",false,2,14000),
      (4,"zl",false,1,18000),
      (5,"win7",false,1,16000) ).toDF("id","name","sex","dept","salary");
    df.createTempView("t_employee2");
    val sql2=
      """
	  |select id,name,salary,dept,
  |count(id) over(partition by dept order by salary desc) as rank,
  |(count(id) over(partition by dept order by salary desc rows between current row and unbounded following) - 1) as low_than_me,
  |avg(salary) over(partition by dept rows between unbounded preceding and unbounded following) as avg_salary,
  |avg(salary) over() as all_avg_salary
  |from t_employee2 t1 order by dept desc
        """.stripMargin;
    spark.sql(sql2).show();
  }

}

SQL解析过程
    println(df.queryExecution);
    println("+"*500);
    println(df.queryExecution.optimizedPlan);