目录

• 一：什么是SparkSQL？
  • （一）SparkSQL简介
  • （二）SparkSQL运行原理
  • （三）SparkSQL特点
• 二：DataFrame
  • （一）什么是DataFrame？
• 补充：Spark中的RDD、DataFrame和DataSet讲解
  • （一）Spark中的模块
  • （二）RDD和DataFrame的区别
• 三：SparkSession
  • （一）SparkSession简介
  • （二）SparkSession实质
  • （三）SparkSession特点
• 四：通过RDD创建DataFrame
  • （一）通过样本类创建（反射）
  • （二）通过SparkSession创建DataFrame
  • （三）通过 json 文件创建DataFrames
• 五：临时视图
  • （一）什么是视图
  • （二）类型
  • （三）创建视图
  • （四）视图查询
  • （五）会话周期
• 六：DataFrame的read和save和savemode
  • （一）数据读取
  • （二）数据保存
  • （三）数据保存模式
• 七：数据集DataSet
  • （一）创建和使用DataSet---使用序列
  • （二）创建和使用DataSet---通过case class作为编码器，将DataFrame转换成DataSet
  • （三）创建和使用DataSet---读取HDFS数据文件

case class People(val name:String,val age:Int)  //可以声明数据类型object WordCount {  def main(args:Array[String]):Unit={    val conf = new SparkConf()    //设置运行模式为本地运行，不然默认是集群模式    //conf.setMaster("local")  //默认是集群模式    //设置任务名    conf.setAppName("WordCount").setMaster("local")    conf.set("spark.default.parallelism","5")    //设置SparkContext,是SparkCore的程序入口    val sc = new SparkContext(conf)    val Sqlsc = new SQLContext(sc)  //根据SparkContext生成SQLContext        val array = Array("mark,14","kitty,23","dasi,45")    val peopleRDD = sc.parallelize(array).map(line=>{    //生成RDD      People(line.split(",")(0),line.split(",")(1).trim().toInt)    })        import Sqlsc.implicits._  //引入全部方法    //将RDD转换成DataFrame    val df = peopleRDD.toDF()      //将DataFrame转换成一个临时的视图    df.createOrReplaceTempView("people")    //使用SQL语句进行查询    Sqlsc.sql("select * from people").show()  }}

object WordCount {  def main(args:Array[String]):Unit={    val conf = new SparkConf()    //设置运行模式为本地运行，不然默认是集群模式    //conf.setMaster("local")  //默认是集群模式    //设置任务名    conf.setAppName("WordCount").setMaster("local")    conf.set("spark.default.parallelism","5")    //设置SparkContext,是SparkCore的程序入口    val sc = new SparkContext(conf)    val Sqlsc = new SQLContext(sc)  //根据SparkContext生成SQLContext        val array = Array("mark,14","kitty,23","dasi,45")    //1.需要将RDD数据映射成Row,需要引入import org.apache.spark.sql.Row    val peopleRDD = sc.parallelize(array).map(line=>{    //生成RDD      val fields = line.split(",")      Row(fields(0),fields(1).trim().toInt)    })        //2.创建StructType定义结构    val st:StructType = StructType(      //字段名，字段类型，是否可以为空      List(  //传参是列表类型，或者使用StructField("name", StringType, true) :: StructField("age", IntegerType, true) :: Nil来构建列表          StructField("name",StringType,true),          StructField("age",IntegerType,true)          )    )        //3.使用SparkSession建立DataFrame    val df = Sqlsc.createDataFrame(peopleRDD,st)    //将DataFrame转换成一个临时的视图    df.createOrReplaceTempView("people")    //使用SQL语句进行查询    Sqlsc.sql("select * from people").show()  }}

[{"name":"dafa","age":12},{"name":"safaw","age":17},{"name":"ge","age":34}]

def main(args:Array[String]):Unit={    val conf = new SparkConf()    //设置运行模式为本地运行，不然默认是集群模式    //conf.setMaster("local")  //默认是集群模式    //设置任务名    conf.setAppName("WordCount").setMaster("local")    //设置SparkContext,是SparkCore的程序入口    val sc = new SparkContext(conf)    val Sqlsc = new SQLContext(sc)  //根据SparkContext生成SQLContext        //通过json数据直接创建DataFrame    val df = Sqlsc.read.json("E:1.json")        //将DataFrame转换成一个临时的视图    df.createOrReplaceTempView("people1")    //使用SQL语句进行查询    Sqlsc.sql("select * from people1").show()  }

val conf = new SparkConf().setAppName("TestDataFrame2").setMaster("local")    val sc = new SparkContext(conf)    val sqlContext = new SQLContext(sc)    //方式一    val df1 = sqlContext.read.json("E:666people.json")    val df2 = sqlContext.read.parquet("E:666users.parquet")    //方式二    val df3 = sqlContext.read.format("json").load("E:666people.json")    val df4 = sqlContext.read.format("parquet").load("E:666users.parquet")    //方式三，默认是parquet格式    val df5 = sqlContext.load("E:666users.parquet")　  //方式四，使用MySQL进行数据源读取    val url = "jdbc:mysql://192.168.123.102:3306/hivedb"    val table = "dbs"    val properties = new Properties()    properties.setProperty("user","root")    properties.setProperty("password","root")    //需要传入Mysql的URL、表明、properties（连接数据库的用户名密码）    val df = sqlContext.read.jdbc(url,table,properties)    df.createOrReplaceTempView("dbs")    sqlContext.sql("select * from dbs").show()

<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-hive -->        <dependency>            <groupId>org.apache.spark</groupId>            <artifactId>spark-hive_2.11</artifactId>            <version>2.3.0</version>        </dependency>

<configuration>        <property>                <name>javax.jdo.option.ConnectionURL</name>                <value>jdbc:mysql://localhost:3306/hivedb?createDatabaseIfNotExist=true</value>                <description>JDBC connect string for a JDBC metastore</description>                <!-- 如果 mysql 和 hive 在同一个服务器节点，那么请更改 hadoop02 为 localhost -->        </property>        <property>                <name>javax.jdo.option.ConnectionDriverName</name>                <value>com.mysql.jdbc.Driver</value>                <description>Driver class name for a JDBC metastore</description>        </property>        <property>                <name>javax.jdo.option.ConnectionUserName</name>                <value>root</value>                <description>username to use against metastore database</description>        </property>        <property>                <name>javax.jdo.option.ConnectionPassword</name>                <value>root</value>        <description>password to use against metastore database</description>        </property>    <property>                <name>hive.metastore.warehouse.dir</name>                <value>/hive/warehouse</value>                <description>hive default warehouse, if nessecory, change it</description>        </property>  </configuration>hive-site.xml配置文件

val conf = new SparkConf().setMaster("local").setAppName(this.getClass.getSimpleName)    val sc = new SparkContext(conf)    val sqlContext = new HiveContext(sc)    sqlContext.sql("select * from myhive.student").show() 

val conf = new SparkConf().setAppName("TestDataFrame2").setMaster("local")    val sc = new SparkContext(conf)    val sqlContext = new SQLContext(sc)    val df1 = sqlContext.read.json("E:666people.json")    //方式一    df1.write.json("E:111")    df1.write.parquet("E:222")    //方式二    df1.write.format("json").save("E:333")    df1.write.format("parquet").save("E:444")    //方式三    df1.write.save("E:555")

df1.write.format("parquet").mode(SaveMode.Ignore).save("E:444")

(1,"Tom")  (2,"Mary")