PySpark（四）PySpark SQL、Catalyst优化器、Spark SQL的执行流程、Spark新特性-CFANZ编程社区

在RDD阶段，程序的执行入口对象是: SparkContext
在Spark 2.0后，推出了SparkSession对象，作为Spark编码的统一入口对象
SparkSession对象可以:
-用于SparkSQL编程作为入口对象
- 用于SparkCore编程，可以通过SparkSession对象中获取到SparkContext

from pyspark.sql import SparkSession
if __name__ == '__main__':
    spark =  SparkSession.builder.appName('lmx').master('local[*]').getOrCreate()
    sc = spark.sparkContext

DataFrame入门

DataFrame构建

1、用RDD进行构建

rdd的结构要求为：[[xx,xx],[xx,xx]]

spark.createDataFrame(rdd,schema=[])

    spark =  SparkSession.builder.appName('lmx').master('local[*]').getOrCreate()
    sc = spark.sparkContext
    rdd = sc.textFile('data/input/sql/people.txt').map(lambda x:x.split(',')).map(lambda x:[x[0],int(x[1])])
    print(rdd.collect())
    # [['Michael', 29], ['Andy', 30], ['Justin', 19]]
    df = spark.createDataFrame(rdd,schema=['name','age'])
    df.printSchema()#打印表结构
    df.show()#打印表
#     root
#     | -- name: string(nullable=true)
#     | -- age: long(nullable=true)
# 
# +-------+---+
# | name | age |
# +-------+---+
# | Michael | 29 |
# | Andy | 30 |
# | Justin | 19 |
# +-------+---+

2、利用StructType进行创建

需要先引入StructType,StringType,IntegerType等构建schema

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType,StringType,IntegerType
if __name__ == '__main__':
    spark =  SparkSession.builder.appName('lmx').master('local[*]').getOrCreate()
    sc = spark.sparkContext
    rdd = sc.textFile('data/input/sql/people.txt').map(lambda x:x.split(',')).map(lambda x:[x[0],int(x[1])])
#构建schema    
schema =StructType().add("name",StringType(),nullable=False).\
        add('age',IntegerType(),nullable=True)
    df = spark.createDataFrame(rdd,schema=schema)
    df.printSchema()
    df.show()

3、toDF将rdd转换为df

下面展示了两种方式

    # 只设定列名，列的数据结构则是内部自己判断
    df = rdd.toDF(['name','age'])
    df.printSchema()
    # root
    # | -- name: string(nullable=true)
    # | -- age: long(nullable=true)
    # 设定列名和数据类型
    schema =StructType().add("name",StringType(),nullable=False).\
        add('age',IntegerType(),nullable=True)
    df = rdd.toDF(schema=schema)
    df.printSchema()
    # root
    # | -- name: string(nullable=false)
    # | -- age: integer(nullable=true)

4、基于pandas构建

    dfp = pd.DataFrame({
        "id":[1,2,3],
        'score':[99,98,100]
    })
    df = spark.createDataFrame(dfp)
    df.printSchema()
    df.show()
    # root
    # | -- id: long(nullable=true)
    # | -- score: long(nullable=true)
    # 
    # +---+-----+
    # | id | score |
    # +---+-----+
    # | 1 | 99 |
    # | 2 | 98 |
    # | 3 | 100 |
    # +---+-----+

5、通过文件读取创造

在读取json和parquet文件时不需要设定schema，因为文件已经自带

而读取csv时，还需要使用.option设定 header等参数

这里说一下parquet文件

DataFrame代码风格

DSL

其实就是用其内置的API处理数据，举例：

    df.select('id','subject').show()
    df.where('subject="语文"').show()
    df.select('id','subject').where('subject="语文"').show()
    df.groupBy('subject').count().show()

API其实跟SQL类似，这里不详细说明了，个人感觉不如直接写SQL语句

SQL

DataFrame的一个强大之处就是我们可以将它看作是一个关系型数据表，然后可以通过在程序中使用spark.sgl0来执行SQL语句查询，结果返回一个DataFrame。如果想使用SQL风格的语法，需要将DataFrame注册成表采用如下的方式:

    df.createTempView('tmp') #创建临时视图
    df.createGlobalTempView('global_tmp')#创建全局试图
    # 全局表: 跨SparkSession对象使用在一个程序内的多个SparkSession中均可调用查询前带上前缀：global_tmp
    df.createOrReplaceTempView('repalce_tmp')#创建临时表，如果存在则替换

然后使用spark.sql的形式书写sql代码

    spark.sql('select * from tmp where subject = "语文"').show()
    spark.sql('select id,score from repalce_tmp where score>90').show()
    spark.sql('select subject,max(score) from global_temp.global_tmp group by subject').show()

SparkSQL Shuffle 分区数目

原因: 在SparkSQL中当Job中产生Shufle时，默认的分区数 spark.sql.shufle,partitions 为200，在实际项目中要合理的设置。
在代码中可以设置：

spark =  SparkSession.builder.appName('lmx').\
master('local[*]').config('spark.sql.shufle,partitions',2).\
getOrCreate()

DataFrame数据写出

统一API：

下面提供两种方法，分别写出为json和csv

    spark.sql(
        'select user_id,avg(score) avg_score from tmp group by user_id order by avg_score desc'
    ).write.mode('overwrite').format('json').save('data/output/1t')

    spark.sql(
        'select user_id,avg(score) avg_score from tmp group by user_id order by avg_score desc'
    ).write.mode('overwrite').format('csv')\
        .option('header',True)\
        .option('sep',';')\
        .save('data/output/csv')

其他的一些方法：

SparkSQL中读取数据和写出数据 - 知乎

不过这里似乎不能自己命名导出的数据文件

Spark UDF

在SparkSQL中，目前仅仅支持UDF函数和UDAF函数，目前Python仅支持UDF

UDF有两种定义方式

举例：

    def double_score(num):
        return 2*num

    udf1 = spark.udf.register('udf_1',double_score,IntegerType())
    # dsl风格
    df.select(udf1(df['score'])).show()
    # sql风格
    df.selectExpr('udf_1(score)').show()
    # sql风格2
    df.createTempView('tmp')
    spark.sql("select udf_1(score) from tmp").show()

    udf2 = F.udf(double_score,IntegerType())
    df.select(udf2(df['score'])).show()

当返回值是数组时，需要定义数组内部数据的数据类型：ArrayType(StringType())

    spark =  SparkSession.builder.appName('lmx').master('local[*]').config('spark.sql.shufle,partitions',2).getOrCreate()
    sc = spark.sparkContext

    rdd=sc.parallelize([['i love you'],['i like you']])
    df = rdd.toDF(['ifo'])
    def func(num):
        return num.split(' ')
    udf = spark.udf.register('udf_sql',func,ArrayType(StringType()))

    # dsl风格
    df.select(udf(df['ifo'])).show()

当返回值是字典时，需要使用StructType()，且定义每个列的名字（需要跟函数返回值的列名一样）和数据类型

    rdd=sc.parallelize([[1],[2],[3],[4],[5]])
    df = rdd.toDF(['ifo'])
    df.show()
    def func(num):
        return {'num':num,'num1':num+10}
    udf = spark.udf.register('udf_sql',func,StructType().\
                             add('num',IntegerType(),nullable=False).\
                             add('num1',IntegerType(),nullable=False))
    df.select(udf(df['ifo'])).show()