spark系列 – spark-sql编程使用

On 2020年10月12日2020年10月12日By yuer

体验过spark RDD编程，我们能感受到几个问题：

RDD最多支持到K-V形式，对编程模型有点束缚。– 不如SQL好用~
RDD中的数据有类型，因此各种算子都需要持续关注模板类型。 — 能不能隐式类型？
RDD没法和Hive表结构天然对应，导致spark清洗数据导出到Hive不方便。–能否支持table形式？

那么这些问题都由spark-sql库来支持：

Spark SQL is a Spark module for structured data processing. Unlike the basic Spark RDD API, the interfaces provided by Spark SQL provide Spark with more information about the structure of both the data and the computation being performed.

spark-sql可以让我们创建table表格，定义表格字段与类型，然后直接SQL操作数据，最后还可以把table写到HIve中去，供后续数仓使用。

同时，spark-sql和spark-core的RDD可以互相转换，下面的例子我们会逐步演示。

下面的示例代码地址：https://github.com/owenliang/spark-sql-demo

引入maven依赖

除了spark-core外，我们额外引入spark-sql库。

后续演示代码用了fastjson库，也引入进来。

    <dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.0.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.3.0</version>
        </dependency>
        <dependency> <!-- Spark dependency -->
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.0.1</version>
            <scope>provided</scope>
        </dependency>
        <!-- https://mvnrepository.com/artifact/com.alibaba/fastjson -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.74</version>
        </dependency>
    </dependencies>

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.12</artifactId>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-client</artifactId>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-sql_2.12</artifactId>

<scope>provided</scope>

</dependency>

<groupId>com.alibaba</groupId>

<artifactId>fastjson</artifactId>

</dependency>

</dependencies>

初始化连接

spark-sql需要使用SparkSession来初始化与spark的连接，这和spark-core是不一样的。

public class Main {
    public static void main(String []args) {
        try {
            // spark sql session
            SparkSession sess = SparkSession.builder().appName("spark-sql-demo").getOrCreate();
            // hdfs
            FileSystem dfs =  FileSystem.get(new Configuration());

public class Main {

public static void main(String []args) {

try {

// spark sql session

SparkSession sess = SparkSession.builder().appName("spark-sql-demo").getOrCreate();

// hdfs

FileSystem dfs = FileSystem.get(new Configuration());

定义Person类

后续我们会定义Person表，所以我们先定义一个Person的Bean类型：

package cc.yuerblog.sql;

import com.alibaba.fastjson.annotation.JSONField;

import java.io.Serializable;

public class PersonBean implements Serializable {
    @JSONField(name="name")
    private String name;
    @JSONField(name="age")
    private int age;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }
}

package cc.yuerblog.sql;

import com.alibaba.fastjson.annotation.JSONField;

import java.io.Serializable;

public class PersonBean implements Serializable {

@JSONField(name="name")

private String name;

@JSONField(name="age")

private int age;

public String getName() {

return name;

}

public void setName(String name) {

this.name = name;

}

public int getAge() {

return age;

}

public void setAge(int age) {

this.age = age;

}

它有name和age两个字段，后续会对应到table的2个列，另外用fastjson注解了一下JSON序列化。

例子1：加载JSON文本为Dataframe

我们接下来在HDFS中生成一个JSON文本文件，每行是一个Person的JSON字符串。

然后我们利用spark-sql的api，将JSON文本加载为一个Dataframe，它就是spark中对table的抽象。

所谓Dataframe，在Java代码中的具体表现就是Dataset<Row>的一个数据集，我们并不会直接看到Dataframe的代码字样。

Dataset是一种更高级的RDD，它可以和JavaRDD互相转化。

Dataset<>的模板参数也支持任意可序列化类型，但是当Dataset用<Row>作为模板参数时，我们把它叫做Dataframe，此时它是一个table，可以直接执行SQL。

所以Dataset是高级的RDD，而Dataset<Row>是更高级的Dataset，我们把它叫做Dataframe。

package cc.yuerblog.sql;

import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.Random;

public class Json2DF {
    public void run(FileSystem dfs, SparkSession sess) throws IOException {
        String path = "/people.txt";

        // 写入1000行JSON测试数据
        FSDataOutputStream out =  dfs.create(new Path(path));
        Random rand = new Random();
        for (int i = 0; i < 1000; i++) {
            PersonBean person = new PersonBean();
            person.setName(String.format("person-%d", i));
            person.setAge(1 + rand.nextInt(100));
            String line = JSONObject.toJSONString(person) + "\n";
            out.write(line.getBytes());
        }
        out.close();

        // 读取JSON文件为一个dataframe（Dataset<Row>）
        Dataset<Row> df = sess.read().json(path);
        // 打印schema
        df.printSchema();
        // 打印数据
        df.show();
    }
}

package cc.yuerblog.sql;

import com.alibaba.fastjson.JSONObject;

import org.apache.hadoop.fs.FSDataOutputStream;

import org.apache.hadoop.fs.FileSystem;

import org.apache.hadoop.fs.Path;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

import java.io.IOException;

import java.util.Random;

public class Json2DF {

public void run(FileSystem dfs, SparkSession sess) throws IOException {

String path = "/people.txt";

// 写入1000行JSON测试数据

FSDataOutputStream out = dfs.create(new Path(path));

Random rand = new Random();

for (int i = 0; i < 1000; i++) {

PersonBean person = new PersonBean();

person.setName(String.format("person-%d", i));

person.setAge(1 + rand.nextInt(100));

String line = JSONObject.toJSONString(person) + "\n";

out.write(line.getBytes());

}

out.close();

// 读取JSON文件为一个dataframe（Dataset<Row>）

Dataset<Row> df = sess.read().json(path);

// 打印schema

df.printSchema();

// 打印数据

df.show();

}

我们打开HDFS文件，随机写入1000个PersonBean的JSON字符串，用换行符分隔，为这些Person生成唯一的name和随机的age。
利用sparkSession.read().json()加载 HDFS文件，可以直接得到Dataset<Row>，也就是DataFrame（简称DF）。
既然DF是表格，我们直接可以打印表格的schema，就像mysql show create table一样，也可以通过show打印一些数据样例。

程序输出如下：

root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)

+---+---------+
|age|     name|
+---+---------+
| 46| person-0|
| 53| person-1|
| 92| person-2|
| 15| person-3|
| 93| person-4|
| 46| person-5|
| 45| person-6|
| 93| person-7|
|  4| person-8|
| 59| person-9|
| 68|person-10|
| 14|person-11|
| 20|person-12|
| 28|person-13|
| 83|person-14|
| 20|person-15|
| 54|person-16|
| 54|person-17|
| 87|person-18|
| 58|person-19|
+---+---------+

root

|-- age: long (nullable = true)

|-- name: string (nullable = true)

+---+---------+

|age| name|

+---+---------+

| 46| person-0|

| 53| person-1|

| 92| person-2|

| 15| person-3|

| 93| person-4|

| 46| person-5|

| 45| person-6|

| 93| person-7|

| 4| person-8|

| 59| person-9|

| 68|person-10|

| 14|person-11|

| 20|person-12|

| 28|person-13|

| 83|person-14|

| 20|person-15|

| 54|person-16|

| 54|person-17|

| 87|person-18|

| 58|person-19|

+---+---------+

可以看到age列是long类型，name列是string类型，这些都是spark-sql自动从json中识别出来的。

这里Row只是一种特殊的类型，它里面存储了表格的列类型和内容，我们还需要通过下面的几个例子来加深理解。

例子2：加载JSON文本，手动转为Dataframe

上个例子中，我们直接利用spark-sql的read().json()方法把json文本解析成了DataFrame。

现在我们把例子1的代码改成手动版，用于加深Dataset理解。

public class Rdd2DF {
    private static class Row2Person implements Function<Row, PersonBean> {
        public PersonBean call(Row row) throws Exception {
            String json = row.getString(0);
            PersonBean person = JSONObject.parseObject(json, PersonBean.class);
            return person;
        }
    }

    public void run(FileSystem dfs, SparkSession sess) throws IOException {
        String path = "/people.txt";

        // 加载Dataframe，每一行就1列
        Dataset<Row> df = sess.read().text(path);

        // Dataframe退化为RDD，方便解析为Person类型
        JavaRDD<PersonBean> personRDD = df.javaRDD().map(new Row2Person());

        // 从RDD转回Dataframe，让Dataframe自动反射java bean的字段
        Dataset<Row> personDF = sess.createDataFrame(personRDD, PersonBean.class);

        // 展示数据
        personDF.show();
    }
}

public class Rdd2DF {

private static class Row2Person implements Function<Row, PersonBean> {

public PersonBean call(Row row) throws Exception {

String json = row.getString(0);

PersonBean person = JSONObject.parseObject(json, PersonBean.class);

return person;

}

public void run(FileSystem dfs, SparkSession sess) throws IOException {

String path = "/people.txt";

// 加载Dataframe，每一行就1列

Dataset<Row> df = sess.read().text(path);

// Dataframe退化为RDD，方便解析为Person类型

JavaRDD<PersonBean> personRDD = df.javaRDD().map(new Row2Person());

// 从RDD转回Dataframe，让Dataframe自动反射java bean的字段

Dataset<Row> personDF = sess.createDataFrame(personRDD, PersonBean.class);

// 展示数据

personDF.show();

}

利用read().text()来加载json文件，虽然返回的也是Dataset<Row>，但是这里的Row里面只有1列原始JSON文本，并没有解析成字段。
利用df的javaRDD()转回JavaRDD，方便进行编程操作。
对javaRDD进行map处理，传入function取出Row的第0列，利用fastjson解析到PersonBean，返回一个RDD<PersonBean>。
最后通过sparkSession的createDataFrame，把RDD<Person>转成Dataset<Row>，这一步spark会自动根据PersonBean的字段来创建Dataframe的列schema。

可见，spark-sql能够从RDD转换为DataFrame，而且schema能够自动从java bean的字段解析得到，还有另外一种方式是createDataFrame时手动传入schema（大家可以文章头部的官方文档）。

例子3：Dataframe执行SQL

说了半天dataframe怎么加载，现在我们看看怎么写SQL：

package cc.yuerblog.sql;

import org.apache.hadoop.fs.FileSystem;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.api.java.function.ReduceFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;

public class DF2View {
    private static class MapFunc implements MapFunction<Row, Integer> {
        public Integer call(Row row) throws Exception {
            Long cnt = row.getAs("cnt");
            return cnt.intValue();
        }
    }
    private static class ReduceFunc implements ReduceFunction<Integer> {
        public Integer call(Integer a, Integer b) throws Exception {
            return a + b;
        }
    }

    public void run(FileSystem dfs, SparkSession sess) throws IOException {
        String path = "/people.txt";

        // 加载dataframe
        Dataset<Row> df = sess.read().json(path);

        // 创建view
        df.createOrReplaceTempView("people");

        // 直接查询SQL
        Dataset<Row> result = sess.sql("select age, count(*) as cnt from people group by age");

        // 打印结果，显示全部行
        result.show();

        // 直接基于Dataset进一步处理，处理结果是Dataset<Integer>
        Dataset<Integer> cntRdd = result.map(new MapFunc(), Encoders.INT());
        Integer total = cntRdd.reduce(new ReduceFunc());
        System.out.println("total=" + total);
    }
}

package cc.yuerblog.sql;

import org.apache.hadoop.fs.FileSystem;

import org.apache.spark.api.java.function.MapFunction;

import org.apache.spark.api.java.function.ReduceFunction;

import org.apache.spark.sql.Dataset;

import org.apache.spark.sql.Encoders;

import org.apache.spark.sql.Row;

import org.apache.spark.sql.SparkSession;

import java.io.IOException;

public class DF2View {

private static class MapFunc implements MapFunction<Row, Integer> {

public Integer call(Row row) throws Exception {

Long cnt = row.getAs("cnt");

return cnt.intValue();

}

private static class ReduceFunc implements ReduceFunction<Integer> {

public Integer call(Integer a, Integer b) throws Exception {

return a + b;

}

public void run(FileSystem dfs, SparkSession sess) throws IOException {

String path = "/people.txt";

// 加载dataframe

Dataset<Row> df = sess.read().json(path);

// 创建view

df.createOrReplaceTempView("people");

// 直接查询SQL

Dataset<Row> result = sess.sql("select age, count(*) as cnt from people group by age");

// 打印结果，显示全部行

result.show();

// 直接基于Dataset进一步处理，处理结果是Dataset<Integer>

Dataset<Integer> cntRdd = result.map(new MapFunc(), Encoders.INT());

Integer total = cntRdd.reduce(new ReduceFunc());

System.out.println("total=" + total);

}

read().json()把json文件加载为dataframe。
在当前sparkSession下，把上述dataframe创建为会话级的临时视图表people。
直接在sparkSession下执行SQL查询people表，就像使用mysql/hive一样，其返回值还是dataframe，里面统计了每个年龄有多少人，包含age和cnt两个字段。
show()打印结果集的前几行。
直接对DataFrame执行map（Dataset是一种高级的RDD），传入MapFunction把每行Row的cnt列返回，结果Dataset是Integer数据类型，你会发现返回的Dataset需要传入的Encoders.INT()序列化方法，这个和RDD不太一样（RDD只要求类型实现Seriablizable接口即可，而Dataset则要求传入一种序列化方法）。
最后再执行一波reduce把所有计数汇总起来。

Dataset是spark-sql的抽象，它比RDD性能更好的原因是，它采用特殊的数据序列化方法，可以在不反序列化数据的情况下直接对某个列进行条件过滤或者聚合统计等，所以如果我们要创建Dataset必须传入Encoders。

Encoders不需要我们自己实现，常见类型都内置，自定义Java bean类型spark-sql会自动反射字段为schema，也支持手动挡配置一个schema出来。

例子4：保存DataFrame到Hive

经过spark清洗的dataframe，最后一般要写入HIVE表。

建议用SQL方式来完成数据的插入，这样就和用HIve没啥区别，不容易坑。

public class DF2Hive {
    public void run(FileSystem dfs, SparkSession sess) throws IOException {
        String path = "/people.txt";
        String hive = "/hive/people";

        // 加载json为dataframe
        Dataset<Row> df = sess.read().json(path);

        // 因为此刻我没安装hive，所以只能保存为hive外表的数据，需要手动去hive建外表
        // 如果安装了hive，可以saveAsTable()保存到hive表，但是会删除旧表所有数据，不太实用。
        df.write().format("orc").mode("overwrite").option("compression", "zlib").save(hive);

        // 因为此刻没装hive，所以也不能用hive表名加载数据，只能直接指定路径
        // 如果装了hive，也可以用table()替换load，可以直接利用hive的元数据来解析表数据
        Dataset<Row> table = sess.read().format("orc").load(hive);
        table.show();

        // 如果hive表已经建立，那么有2个常见方式把数据写到hive表里：
        // 1）数据也可以save写入hive表分区hdfs路径下，然后通过alter table partition让hive元数据发生变更
        // 2）也可以通过sql直接insert到hive表分区下会更加方便，分区元数据会自动变更
    }
}

public class DF2Hive {

public void run(FileSystem dfs, SparkSession sess) throws IOException {

String path = "/people.txt";

String hive = "/hive/people";

// 加载json为dataframe

Dataset<Row> df = sess.read().json(path);

// 因为此刻我没安装hive，所以只能保存为hive外表的数据，需要手动去hive建外表

// 如果安装了hive，可以saveAsTable()保存到hive表，但是会删除旧表所有数据，不太实用。

df.write().format("orc").mode("overwrite").option("compression", "zlib").save(hive);

// 因为此刻没装hive，所以也不能用hive表名加载数据，只能直接指定路径

// 如果装了hive，也可以用table()替换load，可以直接利用hive的元数据来解析表数据

Dataset<Row> table = sess.read().format("orc").load(hive);

table.show();

// 如果hive表已经建立，那么有2个常见方式把数据写到hive表里：

// 1）数据也可以save写入hive表分区hdfs路径下，然后通过alter table partition让hive元数据发生变更

// 2）也可以通过sql直接insert到hive表分区下会更加方便，分区元数据会自动变更

}

加载json为dataframe。
将dataframe覆盖写出到某个hive表的目录下，这里用save方法直接保存到hdfs路径下。
最后再通过load直接加载某个路径下的数据。

show将输出表格内容：

+---+---------+
|age|     name|
+---+---------+
| 83| person-0|
| 96| person-1|
| 96| person-2|
| 34| person-3|
| 93| person-4|
| 64| person-5|
| 25| person-6|
| 33| person-7|
| 91| person-8|
| 72| person-9|
| 90|person-10|
| 74|person-11|
| 68|person-12|
| 66|person-13|
| 93|person-14|
| 92|person-15|
| 79|person-16|
| 36|person-17|
| 84|person-18|
| 51|person-19|

+---+---------+

|age| name|

+---+---------+

| 83| person-0|

| 96| person-1|

| 96| person-2|

| 34| person-3|

| 93| person-4|

| 64| person-5|

| 25| person-6|

| 33| person-7|

| 91| person-8|

| 72| person-9|

| 90|person-10|

| 74|person-11|

| 68|person-12|

| 66|person-13|

| 93|person-14|

| 92|person-15|

| 79|person-16|

| 36|person-17|

| 84|person-18|

| 51|person-19|

实际情况中，建议把dataframe通过createOrReplaceTempView注册为临时表，然后直接sess.sql将数据insert overwrite到hive表中（通常可能涉及到和老表的数据merge，所以SQL是更加实际的）。

无论hive是内表还是外表，如果hive是带分区的，我们都可以直接向hdfs对应分区路径写入数据，然后再去通过hive命令修改hive表的metadata添加该分区信息。

但是上述先写数据再改meta的方法太不实用，实际情况我们还是通过临时表+sql的方式向hive表导入数据更加方便，此时如果需要hive动态分区特性也可以通过配置开启，和hive原生体验基本一致。

总结

spark-sql的意义是把数据清洗后转换为dataframe表格结构，再进一步写入到hive表，这样就可以衔接起ETL和数仓两个部分了。

由于现在还没开始讲hive，所以spark和hive更多的交互留在以后再做更多举例。

如果文章帮助您解决了工作难题，您可以帮我点击屏幕上的任意广告，或者赞助少量费用来支持我的持续创作，谢谢~

引入maven依赖

初始化连接

定义Person类

例子1：加载JSON文本为Dataframe

例子2：加载JSON文本，手动转为Dataframe

例子3：Dataframe执行SQL

例子4：保存DataFrame到Hive

总结

发表回复 取消回复

发表回复取消回复