Java-Spark系列6-Spark SQL编程实战

你的名字 2022-09-12 15:53 229阅读 0赞

### 文章目录 ###

*  一.Spark DataFrame概述
 *   *  1.1 创建DataFrame
     *   *  1.1.1 通过json文件创建DataFrame
         *  1.1.2 通过CSV文件创建DataFrame
         *  1.1.3 通过hive table创建DataFrame
         *  1.1.4 通过jdbc数据源创建DataFrame
 *  二.Spark SQL实战
 *   *  2.1 DataFrame的统计信息
     *  2.2 DataFrame的select操作
     *  2.3 DataFrame对列的操作
     *  2.3 过滤数据
     *  2.4 简单的聚合操作
     *   *  2.4.1 简单聚合
     *  2.5 自定义函数
     *  2.6 表连接
     *   *  2.6.1 内连接
         *  2.6.2 外连接
     *  2.7 排序
     *  2.8 SparkSQL操作文件
 *  参考:

# 一.Spark DataFrame概述 #

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

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

## 1.1 创建DataFrame ##

支持的数据源:

1.  Parquet Files
2.  ORC Files
3.  JSON Files
4.  Hive Tables
5.  JDBC
6.  Avro Files

创建DataFrame的语法:

Dataset<Row> df = spark.read().json("examples/src/main/resources/people.json");

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

import org.apache.spark.sql.SparkSession;
    
    SparkSession spark = SparkSession
      .builder()
      .appName("Java Spark SQL basic example")
      .config("spark.some.config.option", "some-value")
      .getOrCreate();

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

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

Json测试文件:

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

代码:

package org.example;
    
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    public class SparkSQLTest4 {
        public static void main(String[] args){
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest4")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            Dataset<Row> df = spark.read().json("file:///home/pyspark/test.json");
            df.show();
    
            spark.stop();
        }
    
    }

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

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

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

代码:

package org.example;
    
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    public class SparkSQLTest5 {
        public static void main(String[] args){
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest4")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            Dataset<Row> df = spark.read().format("csv").option("header", "true").load("file:///home/pyspark/emp.csv");
            df.show();
    
            spark.stop();
        }
    
    }

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

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

代码:

package org.example;
    
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    public class SparkSQLTest2 {
        public static void main(String[] args){
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest2")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            Dataset<Row> sqlDF = spark.sql("SELECT * FROM test.ods_fact_sale limit 100");
            sqlDF.show();
    
            spark.stop();
        }
    
    }

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

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

代码:

package org.example;
    
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    public class SparkSQLTest3 {
        public static void main(String[] args){
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest3")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            Dataset<Row> jdbcDF = spark.read()
                    .format("jdbc")
                    .option("url", "jdbc:mysql://10.31.1.123:3306/test")
                    .option("dbtable", "(SELECT * FROM EMP) tmp")
                    .option("user", "root")
                    .option("password", "abc123")
                    .load();
    
            jdbcDF.printSchema();
            jdbcDF.show();
    
            spark.stop();
        }
    
    }

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

# 二.Spark SQL实战 #

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

emp
    dept
    bonus
    salgrade

## 2.1 DataFrame的统计信息 ##

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

代码:

package org.example;
    
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    public class SparkSQLTest7 {
        public static void main(String[] args){
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest7")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            spark.sql("use test");
            Dataset<Row> sqlDF = spark.sql("SELECT * FROM emp");
    
            sqlDF.describe().show();
    
            spark.stop();
    
        }
    
    
    }

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

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

## 2.2 DataFrame的select操作 ##

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

代码:

package org.example;
    
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    public class SparkSQLTest8 {
        public static void main(String[] args){
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest8")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            spark.sql("use test");
            Dataset<Row> sqlDF = spark.sql("SELECT * FROM emp");
            sqlDF.select("ename","hiredate").show();
    
            spark.stop();
        }
    }

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

## 2.3 DataFrame对列的操作 ##

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

代码:

package org.example;
    
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    public class SparkSQLTest9 {
        public static void main(String[] args){
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest8")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            spark.sql("use test");
            Dataset<Row> sqlDF = spark.sql("SELECT * FROM emp");
    
            //输出看有哪些列
            System.out.println("\n" + "\n" + "\n");
            for ( String col:sqlDF.columns() ){
                System.out.println(col);
            }
            System.out.println("\n" + "\n" + "\n");
    
            //删除一列
            sqlDF.drop("comm").show();
    
            //新增(或替换)一列
            //sqlDF.withColumn("new_comm", "sal").show();
    
            //给列进行重命名
            sqlDF.withColumnRenamed("comm","comm_new").show();
    
    
            spark.stop();
        }
    }

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

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

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

## 2.3 过滤数据 ##

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

代码:

package org.example;
    
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    public class SparkSQLTest10 {
        public static void main(String[] args) {
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest10")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            spark.sql("use test");
            Dataset<Row> sqlDF = spark.sql("SELECT * FROM emp");
    
            sqlDF.where("comm is not null").show();
            spark.stop();
        }
    }

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

## 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 简单聚合 ###

**代码:**

package org.example;
    
    import org.apache.spark.sql.*;
    
    public class SparkSQLTest11 {
        public static void main(String[] args) {
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest11")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            spark.sql("use test");
            Dataset<Row> sqlDF = spark.sql("SELECT * FROM emp");
    
            sqlDF.groupBy("deptno").agg(functions.avg("sal").alias("avg_sal"),
                                              functions.max("comm").alias("max_comm")).show();
            spark.stop();
    
        }
    
    }

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

## 2.5 自定义函数 ##

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

**代码:**

package org.example;
    
    import org.apache.spark.sql.*;
    import org.apache.spark.sql.api.java.UDF1;
    import org.apache.spark.sql.types.DataTypes;
    
    public class SparkSQLTest12 {
        public static void main(String[] args) {
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest12")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            spark.udf().register("plusOne", new UDF1<Integer, Integer>() {
                @Override
                public Integer call(Integer x) {
                    return x + 1;
                }
            }, DataTypes.IntegerType);
            spark.sql("SELECT plusOne(5)").show();
            spark.stop();
    
        }
    
    
    }

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

## 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 内连接 ###

代码:

package org.example;
    
    import org.apache.spark.sql.*;
    
    public class SparkSQLTest13 {
        public static void main(String[] args) {
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest13")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            spark.sql("use test");
            Dataset<Row> df1 = spark.sql("SELECT * FROM emp");
            Dataset<Row> df2 = spark.sql("SELECT * FROM dept");
    
            Dataset<Row> df3 = df1.join(df2, df1.col("deptno").equalTo(df2.col("deptno")) ,"inner").select(df1.col("empno"),df1.col("ename"),df2.col("dname"),df2.col("loc"));
            df3.show();
            spark.stop();
        }
    
    
    }

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

### 2.6.2 外连接 ###

这里我们使用一个右连接

代码:

package org.example;
    
    import org.apache.spark.sql.*;
    
    public class SparkSQLTest14 {
        public static void main(String[] args) {
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest14")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            spark.sql("use test");
            Dataset<Row> df1 = spark.sql("SELECT * FROM emp");
            Dataset<Row> df2 = spark.sql("SELECT * FROM dept");
    
            Dataset<Row> df3 = df1.join(df2, df1.col("deptno").equalTo(df2.col("deptno")) ,"right").select(df1.col("empno"),df1.col("ename"),df2.col("dname"),df2.col("loc"));
            df3.show();
            spark.stop();
        }
    
    
    }

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

## 2.7 排序 ##

语法:

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

代码:

package org.example;
    
    import org.apache.spark.sql.*;
    
    public class SparkSQLTest15 {
        public static void main(String[] args) {
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest15")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            spark.sql("use test");
            Dataset<Row> df1 = spark.sql("SELECT * FROM emp");
            Dataset<Row> df2 = spark.sql("SELECT * FROM dept");
    
            Dataset<Row> df3 = df1.join(df2, df1.col("deptno").equalTo(df2.col("deptno")) ,"right").select(df1.col("empno"),df1.col("ename"),df2.col("dname"),df2.col("loc"));
            Dataset<Row> df4 = df3.orderBy(df3.col("dname").desc(),df3.col("ename").asc() );
            df4.show();
            spark.stop();
    
        }
    }

测试记录:  
![281d2d5a430253e6b525150671769e44.png][]

## 2.8 SparkSQL操作文件 ##

SparkSession上的sql函数允许应用程序以编程方式运行sql查询，并将结果作为Dataset返回。

代码:

package org.example;
    
    import org.apache.spark.sql.SparkSession;
    import org.apache.spark.sql.Dataset;
    import org.apache.spark.sql.Row;
    
    public class SparkSQLTest16 {
        public static void main(String[] args){
            SparkSession spark = SparkSession
                    .builder()
                    .appName("SparkSQLTest16")
                    .config("spark.some.config.option", "some-value")
                    .getOrCreate();
    
            Dataset<Row> df = spark.read().json("file:///home/pyspark/test.json");
            df.createOrReplaceTempView("people");
    
            spark.sql("select * from people where age = 12").show();
    
            spark.stop();
        }
    
    }

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

# 参考: #

1.  http://spark.apache.org/docs/2.4.2/sql-getting-started.html
2.  http://spark.apache.org/docs/latest/sql-ref-functions-udf-scalar.html
3.  http://spark.apache.org/docs/latest/api/java/index.html

[image.png]: /images/20220828/72d8382c384d47ba868b50ecd4c47156.png
[image.png 1]: /images/20220828/21d0d974d2ac4d0b8190e77323b11631.png
[image.png 2]: /images/20220828/585f8524deec4a1c82af86828237243f.png
[image.png 3]: /images/20220828/8690d49535cc4a52a734a116020f9cff.png
[image.png 4]: /images/20220828/4bc3dd4fadfa453bb71f2963ec95cbe0.png
[image.png 5]: /images/20220828/234dfcb3761d40aa8090c06afa7b3040.png
[image.png 6]: /images/20220828/cecc8e6c39254760ba6d3741d29b4f8e.png
[image.png 7]: /images/20220828/45c3b68c68e44e39a638a3b22ef7a7aa.png
[image.png 8]: /images/20220828/77992120155d46a1bf1de56c5a6e07b5.png
[image.png 9]: /images/20220828/48ad44654c2d40b2ae5f55676608135c.png
[image.png 10]: /images/20220828/15d0a73356604e528bd8d0a7d85668c0.png
[image.png 11]: /images/20220828/621bbdd66190429087bf43cc303cfc5c.png
[image.png 12]: /images/20220828/59edc29f8d0d4b06830902703e5ddb84.png
[image.png 13]: /images/20220828/5acf0b77b4434978ac425a00f4a673ed.png
[image.png 14]: /images/20220828/6ae1258bb53d4843a6f14ffb66026768.png
[281d2d5a430253e6b525150671769e44.png]: /images/20220828/4a367166e437416da19fa433c221e3c0.png
[image.png 15]: /images/20220828/2b65f59460f94b35a0f54c196471e6fc.png