前言
Spark SQL是用于结构化数据处理的Spark模块。它提供了一种称为DataFrame的编程抽象，是由SchemaRDD发展而来。不同于SchemaRDD直接继承RDD，DataFrame自己实现了RDD的绝大多数功能。Spark SQL增加了DataFrame（即带有Schema信息的RDD），使用户可以在Spark SQL中执行SQL语句，数据既可以来自RDD，也可以是Hive、HDFS、Cassandra等外部数据源，还可以是JSON格式的数据。
Spark SQL目前支持Scala、Java、Python三种语言，支持SQL-92规范。
那么根据上篇文章：
PySpark数据分析基础：PySpark基础功能及DataFrame操作基础语法详解
我们知道PySpark可以将DataFrame转换为Spark DataFrame，这为我们python使用Spark SQL提供了实现基础。且在spark3.3.0目录下的pyspark sql可以看到所有函数和类方法：
一、pyspark.sql.SparkSession
基础语法：
class pyspark.sql.SparkSession(sparkContext: pyspark.context.SparkContext, jsparkSession: Optional[py4j.java_gateway.JavaObject] = None, options: Dict[str, Any] = {})
SparkSession是使用Dataset和DataFrame API编程Spark的入口点。
SparkSession可以用于创建DataFrame、将DataFrame注册为表、在表上执行SQL、缓存表和读取parquet文件。要创建SparkSession，需要使用以下生成器模式：
该类可以通过Builder去构建SparkSession：
若对此函数不了解可以去看Spark SQL DataFrame创建一文详解运用与方法和Spark RDD数据操作函数以及转换函数一文详解运用与方法这两篇文章。
使用方法：
spark = SparkSession.builder \ .master("local") \ .appName("Word Count") \ .config("spark.some.config.option", "some-value") \ .getOrCreate()
对标开发Spark的Scala源码：
//暂时保存 用户的sparkContext spark 内部使用 //例如 SQLContext 中 this(SparkSession.builder().sparkContext(sc).getOrCreate()) //用来创建SQLContext实例 private[spark] def sparkContext(sparkContext: SparkContext): Builder = synchronized { userSuppliedContext = Option(sparkContext) this } //设置config到这个类的 options 中去保存，还有各种value其他类型的重载方法 def config(key: String, value: String): Builder = synchronized { options += key -> value this } //设置 spark.app.name def appName(name: String): Builder = config("spark.app.name", name) //把用户在sparkConf中设置的config 加到 这个类的 options 中去 def config(conf: SparkConf): Builder = synchronized { conf.getAll.foreach { case (k, v) => options += k -> v } this } //设置 spark.master 可以是 local、lcoal[*]、local[int] def master(master: String): Builder = config("spark.master", master) //用来检查是否可以支持连接hive 元数据，支持集成hive def enableHiveSupport(): Builder = synchronized { //hiveClassesArePresent 是SparkSession Object的一个方法，用来判断是否包含 //hive的一些支持包（org.apache.spark.sql.hive.HiveSessionStateBuilder，org.apache.hadoop.hive.conf.HiveConf）, //通过ClassForName 反射来判定所需的jar是否存在，自已这里的ClassForName 是spark自己封装的，目的在于尽可能使用启动本线程的 类加载器 //如果所需的集成hive依赖都在的话，hiveClassesArePresent 会返回true if (hiveClassesArePresent) { //在此类Builder的属性options中加入 //catalog的config （spark.sql.catalogImplementation， hive） config(CATALOG_IMPLEMENTATION.key, "hive") } else { throw new IllegalArgumentException( "Unable to instantiate SparkSession with Hive support because " + "Hive classes are not found.") } } //这个方法用来增加扩展点（injection points｜extension points）到 //SparkSessionExtensions里面 def withExtensions(f: SparkSessionExtensions => Unit): Builder = synchronized { f(extensions) this }
那么builder()配置有下面几种：
Method Description
getOrCreate() 获取或创建一个新的SparkSession
enableHiveSupport() 增加Hive支持
appName() 设置application的名字
config() 设置各种配置
master() 加载spark.master设置
那么该函数最直接的使用方法就是创建一个sparkContext：
data = sc.parallelize([1, 2, 3]) data.collect()
[1, 2, 3]
创建转换为RDD。
二、类方法
1.parallelize
该方法为Spark中SparkContext类的原生方法，用于生产一个RDD。
上述已经演示过了。
2.createDataFrame
基础语法
SparkSession.createDataFrame(data,schema=None,samplingRatio=None,verifySchema=True)
功能
从一个RDD、列表或pandas dataframe转换创建为一个Spark DataFrame。
参数说明
data:接受类型为[pyspark.rdd.RDD[Any], Iterable[Any], PandasDataFrameLike]。任何类型的SQL数据表示（Row、tuple、int、boolean等）、列表或pandas.DataFrame的RDD。
schema:接受类型为[pyspark.sql.types.AtomicType, pyspark.sql.types.StructType, str, None] a pyspark.sql.types：数据类型、数据类型字符串或列名列表，默认值为无。数据类型字符串格式等于pyspark.sql.types.DataType.simpleString，除了顶级结构类型可以省略struct<>。
当schema是一列列索引名称时，每一列数据类型将会根据数据类型定义。
当schema为None时，它将尝试从数据中推断模式（列名和类型），判断数据为Row、namedtuple或dict。
当schema为pyspark.sql.types.DataType或为数据类型字符串，它必须与真实数据匹配，否则将在运行时引发异常。如果给定的schema不是pyspark.sql.types.StructType，它将被包装成pyspark.sql.types.StructType作为其唯一字段，字段名将为“value”。每个记录也将被包装成一个元组tuple，也可以将其转换为row。
如果不指定schema，则使用samplingRatio来确定用于模式推理的行的比率。如果samplingRatio为None，则使用第一行。
samplingRatio:接受类型:float, optional。用于推断的行的采样率
verifySchema:根据架构验证每行的数据类型。默认情况下启用。
返回
返回一个pyspark.sql.dataframe.DataFrame。
data参数代码运用：
pd_df=pd.DataFrame( {'name':['id1','id2','id3','id4'], 'old':[21,23,22,35], 'city':['杭州','北京','南昌','上海'] }, index=[1,2,3,4]) spark.createDataFrame(pd_df).collect()
simple=[('杭州','40')] spark.createDataFrame(simple,['city','temperature']).collect()
simple_dict=[{'name':'id1','old':21}] spark.createDataFrame(simple_dict).collect()
[Row(name='id1', old=21)]
rdd = sc.parallelize(simple) spark.createDataFrame(rdd).collect()
[Row(_1='杭州', _2='40')]
schema参数代码运用：
simple=[('杭州',40)] rdd = sc.parallelize(simple) spark.createDataFrame(rdd, "city:string,temperatur:int").collect()
[Row(city='杭州', temperatur=40)]
3.getActiveSession
基础语法：
classmethod SparkSession.getActiveSession()
功能：
返回通过builder生成的当前线程的活动SparkSession
代码示例
s = SparkSession.getActiveSession() simple=[('杭州',40)] rdd = s.sparkContext.parallelize(simple) df = s.createDataFrame(rdd, ['city', 'temperatur']) df.select("city").collect()
[Row(city='杭州')]
4.newSession
基础语法：
SparkSession.newSession()
功能：
将新的SparkSession作为新会话返回，该会话具有单独的SQLConf、注册的临时视图和UDF，但共享SparkContext和表缓存。
重新返回一个新的newSparkSession，作表数据对比时可用。
5.range
基础语法：
SparkSession.range(start,end= None, step: int = 1, numPartitions: = None)
功能：
使用单个pyspark.sql.types.LongType列名为id，包含从开始到结束（独占）范围内的元素，步长值为step。
参数说明：
start:类型[int]，开始值
end:类型[int]，结束值
step:类型[int]，步长
numPartitions:类型[int]，DataFrame分区数
代码示例：
spark.range(1, 100, 20).collect()
[Row(id=1), Row(id=21), Row(id=41), Row(id=61), Row(id=81)]
如果只指定了一个参数，则将其用作结束值，默认起始值为0。
spark.range(5).collect()
[Row(id=0), Row(id=1), Row(id=2), Row(id=3), Row(id=4)]
6.sql
基础语法：
SparkSession.sql(sqlQuery: str, **kwargs: Any)
功能：
返回表示给定查询结果的DataFrame。当指定kwargs时，此方法使用Python标准格式化程序格式化给定字符串。
参数说明：
- sqlQuery：接受类型[str]，SQL查询字符串。
- kwargs:接受类型[dict]，用户想要设置的、可以在查询中引用的其他变量。
代码示例：
spark.sql("SELECT current_date() FROM range(5)").show()
spark.sql( "SELECT * FROM range(10) WHERE id > {bound1} AND id < {bound2}", bound1=7, bound2=9 ).show()
mydf = spark.range(10) spark.sql( "SELECT {col} FROM {mydf} WHERE id IN {x}", col=mydf.id, mydf=mydf, x=tuple(range(4))).show()
复杂运用：
spark.sql(''' SELECT m1.a, m2.b FROM {table1} m1 INNER JOIN {table2} m2 ON m1.key = m2.key ORDER BY m1.a, m2.b''', table1=spark.createDataFrame([(1, "a"), (2, "b")], ["a", "key"]), table2=spark.createDataFrame([(3, "a"), (4, "b"), (5, "b")], ["b", "key"])).show()
此外，还可以使用DataFrame中的class:Column进行查询。
mydf = spark.createDataFrame([(1, 4), (2, 4), (3, 6)], ["A", "B"]) spark.sql("SELECT {df.A}, {df[B]} FROM {df}", df=mydf).show()
7.table
基础语法
SparkSession.table(tableName: str)
功能：
将指定的表作为DataFrame返回。
