pyspark系列6-Spark SQL编程实战

小鱼儿 2023-01-18 12:26 250阅读 0赞

# 一.Spark DataFrame概述 #

从上一篇博客，我们可以知道因为Python是弱类型，所以PySpark SQL的数据抽象就只有DataFrame,这里我们再来复习一下DataFrame。

在Spark语义中，DtatFrame是一个分布式的行集合，可以想象为一个关系型数据库的表，或一个带有列头的Excel表格。它和RDD一样，有这样一些特点:

1.  Immuatable: 一旦RDD、DataFrame被创建，就不能更改，只能通过tranformation生成新的RDD、DataFrame
2.  Lazy Evaluations: 只有action才会出发Transformation的执行。
3.  Distributed: DataFrame和RDD一样都是分布式的。

## 1.1 创建DataFrame ##

创建DataFrame的方式:

1.  pandas dataframe
2.  list
3.  RDD

支持的数据源:

1.  csv
2.  json
3.  parquet
4.  orc
5.  jdbc

创建DataFrame的语法:

-- Spark 1.0 版本
    schemaPeople = sqlContext.createDataFrame(people) 
    -- Spark 2.0 版本
    schemaPeople = spark.createDataFrame(people)

**Spark SQL的起点: SparkSession**  
代码:

from pyspark.sql import SparkSession
    
    spark = SparkSession \
        .builder \
        .appName("Python Spark SQL basic example") \
        .config("spark.some.config.option", "some-value") \
        .getOrCreate()

使用SparkSession，应用程序可以从现有的RDD、Hive表或Spark数据源中创建DataFrames。

### 1.1.1 通过json文件创建DataFrame ###

测试多行的数据会报错:

pyspark.sql.utils.AnalysisException: u'Since Spark 2.3, the queries from raw JSON/CSV files are disallowed when the\nreferenced columns only include the internal corrupt record column\n(named _corrupt_record by default). For example:\nspark.read.schema(schema).json(file).filter($"_corrupt_record".isNotNull).count()\nand spark.read.schema(schema).json(file).select("_corrupt_record").show().\nInstead, you can cache or save the parsed results and then send the same query.\nFor example, val df = spark.read.schema(schema).json(file).cache() and then\ndf.filter($"_corrupt_record".isNotNull).count().;'
    21/04/25 14:53:17 INFO spark.SparkContext: Invoking stop() from shutdown hook

所以多行的json文件调整为单行  
Json测试文件:

{"name": "Michael",  "age": 12}
    {"name": "Andy",  "age": 13}
    {"name": "Justin",  "age": 8}

代码:

测试记录:  
![image.png][]

### 1.1.2 通过CSV文件创建DataFrame ###

csv测试文件:  
![image.png][image.png 1]

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    df = spark.read.format("csv"). \
         option("header", "true"). \
         load("file:///home/pyspark/emp.csv")
    
    df.show()
    
    # 关闭spark会话
    spark.stop()

测试记录:  
![image.png][image.png 2]

### 1.1.3 通过已存在的RDD创建DataFrame ###

我们可以通过已经存在RDD创建DataFrame

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    sc = spark.sparkContext
    
    list1 = [('Ankit',25),('Jalfaizy',22),('saurabh',20),('Bala',26)]
    rdd1 = sc.parallelize(list1)
    
    df1 = spark.createDataFrame(rdd1)
    df1.show()

测试记录:  
![image.png][image.png 3]

### 1.1.4 通过hive table创建DataFrame ###

代码:

测试记录:  
![image.png][image.png 4]

### 1.1.5 通过jdbc数据源创建DataFrame ###

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    df1=spark.read.format("jdbc").options(url="jdbc:mysql://10.31.1.123:3306/test",
                                           driver="com.mysql.jdbc.Driver",
                                           dbtable="(SELECT * FROM EMP) tmp",
                                           user="root",
                                           password="abc123").load()
    
    df1.show()

测试记录:  
![image.png][image.png 5]

# 二.Spark SQL实战 #

我们选用经典scoot用户下的4张表来模拟Spark SQL实战：

emp
    dept
    bonus
    salgrade

## 2.1 DataFrame的统计信息 ##

生成DataFrame的时候会保留统计信息，有点类似关系型数据库的统计信息

代码:

**测试记录：**  
从下图可以看出，DataFrame给每一列都做了统计信息。

1.  count 是列不为空的总数
2.  mean 平均值
3.  stddev 标准偏差
4.  min 最小值
5.  max 最大值  
    ![image.png][image.png 6]

## 2.2 DataFrame的select操作 ##

有些应用场景，我们只需要DataFrame的部分列，此时可以通过select实现:

代码:

测试记录:  
![image.png][image.png 7]

## 2.3 DataFrame对列的操作 ##

有些应用场景，我们需要对列进行别名、新增列、删除列等操作。

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    # spark.sql执行默认是取值Hive表，类型是DataFrame类型
    spark.sql("use test")
    df1 = spark.sql("select * from emp")
    
    # 显示DataFrame有哪些列
    print(df1.columns)
    
    # 删除一列
    df1.drop('comm').show()
    
    # 新增(或替换)一列
    df1.withColumn('new_comm',df1.sal * 0.25).show()
    
    # 给列进行重命名
    df1.withColumnRenamed('comm','comm_new').show()

**测试记录:**  
显示列的信息:  
![image.png][image.png 8]

删除一列:  
![image.png][image.png 9]

新增一列:  
![image.png][image.png 10]

替换列名:  
![image.png][image.png 11]

## 2.3 过滤数据 ##

过滤数据用的是filter，其实也可以用where，where是filter的别名

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    # spark.sql执行默认是取值Hive表，类型是DataFrame类型
    spark.sql("use test")
    df1 = spark.sql("select * from emp")
    
    # 过滤掉奖金为空的数据
    df1.where('comm is not null').show()

测试记录:  
![image.png][image.png 12]

## 2.4 简单的聚合操作 ##

常用的聚合操作:

<table> 
 <thead> 
  <tr> 
   <th>操作</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>avg/mean</td> 
   <td>平均值</td> 
  </tr> 
  <tr> 
   <td>count</td> 
   <td>统计个数</td> 
  </tr> 
  <tr> 
   <td>countDistinct</td> 
   <td>统计唯一的个数</td> 
  </tr> 
  <tr> 
   <td>max</td> 
   <td>求最大值</td> 
  </tr> 
  <tr> 
   <td>min</td> 
   <td>求最小值</td> 
  </tr> 
  <tr> 
   <td>sum</td> 
   <td>求和</td> 
  </tr> 
  <tr> 
   <td>sumDistinct</td> 
   <td>统计唯一值的合计</td> 
  </tr> 
  <tr> 
   <td>skewness</td> 
   <td>偏态</td> 
  </tr> 
  <tr> 
   <td>stddev</td> 
   <td>标准偏差</td> 
  </tr> 
 </tbody> 
</table>

### 2.4.1 简单聚合 ###

**代码:**

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    # spark.sql执行默认是取值Hive表，类型是DataFrame类型
    spark.sql("use test")
    df1 = spark.sql("select * from emp")
    
    # 过滤掉奖金为空的数据
    df1.groupby('deptno').agg({ 'empno':'count','sal':'min','comm':'avg','hiredate':'max'}).show()

测试记录:  
![image.png][image.png 13]

### 2.4.2 自定义聚合 ###

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    import pyspark.sql.functions as fn
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    # spark.sql执行默认是取值Hive表，类型是DataFrame类型
    spark.sql("use test")
    df1 = spark.sql("select * from emp")
    
    # 过滤掉奖金为空的数据
    #df1.groupby('deptno').agg({'empno':'count','sal':'min','comm':'avg','hiredate':'max'}).show()
    
    df1.groupby('deptno').agg(fn.count('empno').alias('emp_count'),
                               fn.min('sal').alias('sal_min')
                              ).show()

测试记录:  
![image.png][image.png 14]

## 2.5 自定义函数 ##

一些比较复杂的场景，我们希望使用自定义函数来实现。

**代码:**

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    import pyspark.sql.functions as fn
    from pyspark.sql.functions import udf
    from pyspark.sql.types import StringType
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    # spark.sql执行默认是取值Hive表，类型是DataFrame类型
    spark.sql("use test")
    df1 = spark.sql("select * from emp")
    
    def fun_double(x):
        return x*2
    
    udf_double = udf(fun_double,StringType())
    
    
    df2 = df1.select('sal',udf_double(df1['sal']).alias('new_sal'))
    
    df2.show()

**测试记录:**  
![image.png][image.png 15]

## 2.6 表连接 ##

语法:

DataFrame.join(other, on=None, how=None)
    
    other            需要连接的DataFrame
    on                str, list or Column, 可选项
    how             str, 可选项
                       default inner. Must be one of: inner, cross, outer, full, fullouter, full_outer, left, leftouter, left_outer, right, rightouter, right_outer, semi, leftsemi, left_semi, anti, leftanti and left_anti

### 2.6.1 内连接 ###

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    # spark.sql执行默认是取值Hive表，类型是DataFrame类型
    spark.sql("use test")
    df1 = spark.sql("select * from emp")
    df2 = spark.sql("select * from dept")
    
    df3 = df1.join(df2,df1.deptno == df2.deptno,'inner').select(df1.empno,df1.ename,df2.dname,df2.loc).show()

测试记录:  
![image.png][image.png 16]

### 2.6.2 外连接 ###

这里我们使用一个右连接

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    # spark.sql执行默认是取值Hive表，类型是DataFrame类型
    spark.sql("use test")
    df1 = spark.sql("select * from emp")
    df2 = spark.sql("select * from dept")
    
    df3 = df1.join(df2,df1.deptno == df2.deptno,'right').select(df1.empno,df1.ename,df2.dname,df2.loc).show()

测试记录:  
![image.png][image.png 17]

## 2.7 排序 ##

语法:

DataFrame.orderBy(*cols, **kwargs)
    -- 返回按指定列排序的新DataFrame
    
    参数:      ascending   bool or list,可选项
                  布尔值或布尔值列表(默认为True)。排序升序与降序。为多个排序顺序指定列表。如果指定了列表，则列表的长度必须等于cols的长度。

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark.sql import SparkSession
    from pyspark.sql.functions import desc,asc
    
    # 创建一个连接
    spark = SparkSession. \
            Builder(). \
            appName('sql'). \
            master('local'). \
            getOrCreate()
    
    # spark.sql执行默认是取值Hive表，类型是DataFrame类型
    spark.sql("use test")
    df1 = spark.sql("select * from emp")
    df2 = spark.sql("select * from dept")
    
    df3 = df1.join(df2,df1.deptno == df2.deptno,'right').select(df1.empno,df1.ename,df2.dname,df2.loc)
    
    df4 = df3.orderBy(desc("dname"),asc("ename"))
    
    df4.show()

测试记录:  
![image.png][image.png 18]

# 参考: #

1.  http://spark.apache.org/docs/latest/sql-getting-started.html
2.  http://spark.apache.org/docs/latest/api/python/reference/pyspark.sql.html\#dataframe-apis

[image.png]: /images/20221021/b1cde9a14ef5479b941124f10c17ebc7.png
[image.png 1]: /images/20221021/aa8c375c69ed441684a5f660d3ce41b9.png
[image.png 2]: /images/20221021/71aa262d797b46b0808af4dc919a6749.png
[image.png 3]: /images/20221021/f9e3440ad916454bbbce41216154a067.png
[image.png 4]: /images/20221021/2d5ea5a0f0d54a2e8e9dcc2b7e3fde67.png
[image.png 5]: /images/20221021/778dfe530db24c848436c90b65d12cdc.png
[image.png 6]: /images/20221021/1f3d338b1cf74d90b5b606f5c0acdaf0.png
[image.png 7]: /images/20221021/40185ab61b5547df93a62f4f2a6c0eec.png
[image.png 8]: /images/20221021/0b4dd6ea8d864138b660c52490d2ec04.png
[image.png 9]: /images/20221021/c993667f418b4056a02a09fabcfc7d35.png
[image.png 10]: /images/20221021/13418b28b7614e08bc9bfa60bef38f58.png
[image.png 11]: /images/20221021/fad871d54a1f4205bd2aa4f64b8f6cfb.png
[image.png 12]: /images/20221021/d03286d77d7e423e9932078714df0ff4.png
[image.png 13]: /images/20221021/4d3351f0cc22400caa65fe452015f69e.png
[image.png 14]: /images/20221021/9d95f5530ee64b81b4abcc7a0ee628a8.png
[image.png 15]: /images/20221021/1736e3f7aa514c5c8f3135bacd66bd12.png
[image.png 16]: /images/20221021/078e101743a04767b191b568ea28cb21.png
[image.png 17]: /images/20221021/fdc6443b8729445c86513b98a20e6054.png
[image.png 18]: /images/20221021/6e1e792f3be34cfbae18aa144395b89a.png