一、概述

(1) 在Spark1.6 中我们使用的叫Hive on spark，主要是依赖hive生成spark程序，有两个核心组件SQLcontext和HiveContext。

这是Spark 1.x 版本的语法

//set up the spark configuration and create contexts
 val sparkConf = new SparkConf().setAppName("SparkSessionZipsExample").setMaster("local")

// your handle to SparkContext to access other context like SQLContext
 val sc = new SparkContext(sparkConf).set("spark.some.config.option", "some-value")
 val sqlContext = new org.apache.spark.sql.SQLContext(sc)

（2）Spark2.0中我们使用的就是sparkSQL，是后继的全新产品，解除了对Hive的依赖。

从Spark2.0以上的版本开始，spark是使用全新的SparkSession接口代替Spark1.6 中的SQLcontext和HiveContext 来实现对数据的加载、转换、处理等工作，并且实现了SQLcontext和HiveContext的所有功能。

在新版本中并不需要之前那么繁琐的创建很多对象，只需要创建一个SparkSession对象即可。SparkSession支持从不同的数据源加载数据，并把数据转换成DataFrame，并支持把DataFrame转换成SQLContext自身中的表。然后使用SQL语句来操作数据，也提供了HiveQL以及其他依赖于Hive的功能支持。

二、创建SparkSession

SparkSession 是 Spark SQL 的入口。使用 Dataset 或者 Dataframe 编写 Spark SQL 应用的时候，第一个要创建的对象就是 SparkSession。Builder 是 SparkSession 的构造器。通过 Builder, 可以添加各种配置，并通过 stop 函数来停止 SparkSession。

Builder 的方法如下：

import org.apache.spark.sql.SparkSession
val spark: SparkSession = SparkSession.builder
 .appName("My Spark Application")  //设置 application 的名字
 .master("local[*]")             
 .enableHiveSupport() //增加支持 hive Support      
 .config("spark.sql.warehouse.dir", "target/spark-warehouse") //设置各种配置
 .getOrCreate     //获取或者新建一个 sparkSession

（1）设置参数

创建SparkSession之后可以通过 spark.conf.set 来设置运行参数

//配置设置
 spark.conf.set("spark.sql.shuffle.partitions", 6)
 spark.conf.set("spark.executor.memory", "2g")

 //获取配置，可以使用Scala的迭代器来读取configMap中的数据。 
 val configMap:Map[String, String] = spark.conf.getAll()

（2）读取元数据

如果需要读取元数据(catalog)，可以通过SparkSession来获取。

 spark.catalog.listTables.show(false)
 spark.catalog.listDatabases.show(false)

这里返回的都是Dataset，所以可以根据需要再使用Dataset API来读取

catalog 和 schema 是两个不同的概念
Catalog是目录的意思，从数据库方向说，相当于就是所有数据库的集合;
Schema是模式的意思, 从数据库方向说, 类似Catelog下的某一个数据库;

（3）创建Dataset和Dataframe

通过SparkSession来创建Dataset和Dataframe有多种方法。

通过range()方法来创建dataset
通过createDataFrame()来创建dataframe。

// create a Dataset using spark.range starting from 5 to 100,
// with increments of 5
val numDS = spark.range(5, 100, 5)//创建dataset

// reverse the order and display first 5 items
numDS.orderBy(desc("id")).show(5)

//compute descriptive stats and display them
numDs.describe().show()

// create a DataFrame using spark.createDataFrame from a List or Seq
val langPercentDF = spark.createDataFrame( List( ("Scala", 35), 
    ("Python", 30), ("R", 15), ("Java", 20)) )//创建dataframe

//rename the columns
val lpDF = langPercentDF.withColumnRenamed("_1", "language").
    withColumnRenamed("_2", "percent")

//order the DataFrame in descending order of percentage
lpDF.orderBy(desc("percent")).show(false)

（4）读取数据

可以用SparkSession读取JSON、CSV、TXT 和 parquet表。

import spark.implicits //使RDD转化为DataFrame以及后续SQL操作
//读取JSON文件，生成DataFrame
val jsonFile = args(0)
val zipsDF = spark.read.json(jsonFile)

（5）使用SparkSQL

借助SparkSession用户可以像SQLContext一样使用Spark SQL的全部功能。

zipsDF.createOrReplaceTempView("zips_table")//对上面的dataframe创建一个表
zipsDF.cache()//缓存表

val resultsDF = spark.sql("SELECT city, pop, state, zip FROM zips_table")

//对表调用SQL语句
resultsDF.show(10)//展示结果

（6）存储/读取Hive表

下面的代码演示了通过SparkSession来创建Hive表并进行查询的方法。

//drop the table if exists to get around existing table error
 spark.sql("DROP TABLE IF EXISTS zips_hive_table")

 //save as a hive table
 spark.table("zips_table").write.saveAsTable("zips_hive_table")

 //make a similar query against the hive table
 val resultsHiveDF = spark.sql("SELECT city, pop, state, 
        zip FROM zips_hive_table WHERE pop > 40000")

 resultsHiveDF.show(10)

ITPub博客

SparkSession和sparkSQL的区别

一、概述

二、创建SparkSession

最新文章