Spark大数据处理讲课笔记4.2 Spark SQL数据源 - 基本操作

目录

零、本讲学习目标

一、基本操作

二、默认数据源

（一）默认数据源Parquet

（二）案例演示读取Parquet文件

1、在Spark Shell中演示

2、通过Scala程序演示

三、手动指定数据源

（一）format()与option()方法概述

（二）案例演示读取不同数据源

1、读取房源csv文件

2、读取json，保存为parquet

3、读取jdbc数据源，保存为json文件

四、数据写入模式

（一）mode()方法

（二）枚举类SaveMode

（三）案例演示不同写入模式

五、分区自动推断

（一）分区自动推断概述

（二）分区自动推断演示

1、建四个文件

2、读取表数据

3、输出Schema信息

4、显示数据帧内容

（三）分区自动推断注意事项

零、本讲学习目标

1. 学会使用默认数据源
2. 学会手动指定数据源
3. 理解数据写入模式
4. 掌握分区自动推断

Spark SQL支持通过DataFrame接口对各种数据源进行操作。DataFrame可以使用相关转换算子进行操作，也可以用于创建临时视图。将DataFrame注册为临时视图可以对其中的数据使用SQL查询。

一、基本操作

• Spark SQL提供了两个常用的加载数据和写入数据的方法：load()方法和save()方法。load()方法可以加载外部数据源为一个DataFrame，save()方法可以将一个DataFrame写入指定的数据源。

二、默认数据源

（一）默认数据源Parquet

• 默认情况下，load()方法和save()方法只支持Parquet格式的文件，Parquet文件是以二进制方式存储数据的，因此不可以直接读取，文件中包括该文件的实际数据和Schema信息，也可以在配置文件中通过参数spark.sql.sources.default对默认文件格式进行更改。Spark SQL可以很容易地读取Parquet文件并将其数据转为DataFrame数据集。

（二）案例演示读取Parquet文件

• 将数据文件users.parquet上传到master虚拟机/home
• 将数据文件users.parquet上传到HDFS的 /datasource/input目录

1、在Spark Shell中演示

• 启动Spark Shell，执行命令：spark-shell --master spark://master:7077
• 加载parquet文件，返回数据帧
• 执行命令：val userdf = spark.read.load("hdfs://master:9000/datasource/input/users.parquet")
• 执行命令：userdf.show()，查看数据帧内容
• 执行命令：userdf.select("name", "favorite_color").write.save("hdfs://master:9000/datasource/output")，对数据帧指定列进行查询，查询结果依然是数据帧，然后通过save()方法写入HDFS指定目录
• 查看HDFS上的输出结果
• 除了使用select()方法查询外，也可以使用SparkSession对象的sql()方法执行SQL语句进行查询，该方法的返回结果仍然是一个DataFrame。
• 基于数据帧创建临时视图，执行命令：userdf.createTempView("t_user")
• 执行SQL查询，将结果写入HDFS，执行命令：spark.sql("select name, favorite_color from t_user").write.save("hdfs://master:9000/result2")
• 查看HDFS上的输出结果

课堂练习1、将4.1节的student.txt文件转换成student.parquet，保存到HDFS的/datasource/input目录

• 解决思路：将student.txt转成studentdf，利用数据帧的save()方法保存到/datasource/output3目录，然后将文件更名复制到/datasource/input目录

• 得到学生数据帧

• 将学生数据帧保存为parquet文件

• 查看生成的parquet文件

• 复制parquet文件到/datasource/input目录

课堂练习2、读取student.parquet文件得到学生数据帧，并显示数据帧内容

• 执行命令：val studentDF = spark.read.load("hdfs://master:9000/datasource/input/student.parquet")
• 执行命令：studentDF.show
• 

2、通过Scala程序演示

• 创建Maven项目 - SparkSQLDemo
• 
• 在pom.xml文件里添加依赖与插件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>net.huawei.sql</groupId>
    <artifactId>SparkSQLDemo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.12.15</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.1.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.1.3</version>
        </dependency>
    </dependencies>
    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
    </build>
</project>

• 在resources目录里添加HFDS配置文件
• 

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <property>
        <description>only config in clients</description>
        <name>dfs.client.use.datanode.hostname</name>
        <value>true</value>
    </property>
</configuration>

• 在resources目录里添加日志属性文件
• 

log4j.rootLogger=ERROR, stdout, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spark.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

• 创建net.hw.sparksql包，在包里创建ReadParquet对象
  

package net.hw.sparksql

import org.apache.spark.sql.SparkSession

/**
 * 功能：Parquet数据源
 * 作者：华卫
 * 日期：2022年05月01日
 */
object ReadParquet {
  def main(args: Array[String]): Unit = {
    // 本地调试必须设置，否则会报Permission Denied错误
    System.setProperty("HADOOP_USER_NAME", "root")
    // 创建或得到SparkSession
    val spark = SparkSession.builder()
      .appName("ReadParquet")
      .master("local[*]")
      .getOrCreate()
    // 加载parquet文件，返回数据帧
    val usersdf = spark.read.load("hdfs://master:9000/input/users.parquet")
    // 显示数据帧内容
    usersdf.show()
    // 查询DataFrame中指定列，结果写入HDFS
    usersdf.select("name","favorite_color")
      .write.save("hdfs://master:9000/result3")
  }
}

• 运行程序，查看控制台结果
• 在HDFS查看输出结果

三、手动指定数据源

（一）format()与option()方法概述

• 使用format()方法可以手动指定数据源。数据源需要使用完全限定名（例如org.apache.spark.sql.parquet），但对于Spark SQL的内置数据源，也可以使用它们的缩写名（JSON、Parquet、JDBC、ORC、Libsvm、CSV、Text）。
• 通过手动指定数据源，可以将DataFrame数据集保存为不同的文件格式或者在不同的文件格式之间转换。
• 在指定数据源的同时，可以使用option()方法向指定的数据源传递所需参数。例如，向JDBC数据源传递账号、密码等参数。

（二）案例演示读取不同数据源

1、读取房源csv文件

• 查看HDFS上/input目录里的house.csv文件
• 在spark shell里，执行命令：val house_csv_df = spark.read.format("csv").load("hdfs://master:9000/input/house.csv")，读取房源csv文件，得到房源数据帧
• 执行命令：house_csv_df.show()，查看房源数据帧内容
• 大家可以看到，house.csv文件第一行是字段名列表，但是转成数据帧之后，却成了第一条记录，这样显然是不合理的，怎么办呢？就需要用到option()方法来传递参数，告诉Spark第一行是表头header，而不是表记录。
• 执行命令：val house_csv_df = spark.read.format("csv").option("header", "true").load("hdfs://master:9000/input/house.csv")
• 执行命令：house_csv_df.show()，查看房源数据帧内容

2、读取json，保存为parquet

• 将people.json上传到HDFS的/input目录
• 执行命令：val peopledf = spark.read.format("json").load("hdfs://master:9000/input/people.json")
• 执行命令：peopledf.show()
• 执行命令：peopledf.select("name", "age").write.format("parquet").save("hdfs://master:9000/result4")
• 查看生成的parquet文件

3、读取jdbc数据源，保存为json文件

• 查看student数据库里的t_user表
• 执行命令

val userdf = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://master:3306/student")
  .option("driver", "com.mysql.jdbc.Driver")
  .option("dbtable", "t_user")  
  .option("user", "root")  
  .option("password", "903213")
  .load()

• 报错，找不到数据库驱动程序com.mysql.jdbc.Driver
• 解决问题，将数据库驱动程序拷贝到$SPARK_HOME/jars目录
• 将数据驱动程序分发到slave1和slave2虚拟机
• 执行命令

val userdf = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://master:3306/student")
  .option("driver", "com.mysql.jdbc.Driver")
  .option("dbtable", "t_user")  
  .option("user", "root")  
  .option("password", "903213")
  .load()

• 加载jdbc数据源成功，但是有个警告，需要通过设置useSSL=false来消除
• 执行命令

val userdf = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://master:3306/student?useSSL=false")
  .option("driver", "com.mysql.jdbc.Driver")
  .option("dbtable", "t_user")  
  .option("user", "root")  
  .option("password", "903213")
  .load()

文章来源地址https://www.toymoban.com/news/detail-491654.html

• 执行命令：userdf.show()
• 执行命令：userdf.write.format("json").save("hdfs://master:9000/result5")
• 在虚拟机slave1查看生成的json文件，执行命令：hdfs dfs -cat /result5/*

四、数据写入模式

（一）mode()方法

• 在写入数据时，可以使用mode()方法指定如何处理已经存在的数据，该方法的参数是一个枚举类SaveMode。
• 使用SaveMode类，需要import org.apache.spark.sql.SaveMode;

（二）枚举类SaveMode

• SaveMode.ErrorIfExists：默认值。当向数据源写入一个DataFrame时，如果数据已经存在，就会抛出异常。
• SaveMode.Append：当向数据源写入一个DataFrame时，如果数据或表已经存在，会在原有的基础上进行追加。
• SaveMode.Overwrite：当向数据源写入一个DataFrame时，如果数据或表已经存在，就会将其覆盖（包括数据或表的Schema）。
• SaveMode.Ignore：当向数据源写入一个DataFrame时，如果数据或表已经存在，就不会写入内容，类似SQL中的CREATE TABLE IF NOT EXISTS。

（三）案例演示不同写入模式

• 查看数据源：people.json
• 查询该文件name里，采用覆盖模式写入/result，/result目录里本来有东西的
• 执行命令：val peopledf = spark.read.format("json").load("hdfs://master:9000/input/people.json")
• 导入SaveMode类，执行命令：peopledf.select("name").write.mode(SaveMode.Overwrite).format("json").save("hdfs://master:9000/result")
• 在slave1虚拟机上查看生成的json文件
• 查询age列，以追加模式写入HDFS的/result目录，执行命令：peopledf.select("age").write.mode(SaveMode.Append).format("json").save("hdfs://master:9000/result")
• 在slave1虚拟机上查看追加生成的json文件

五、分区自动推断

（一）分区自动推断概述

• 表分区是Hive等系统中常用的优化查询效率的方法（Spark SQL的表分区与Hive的表分区类似）。在分区表中，数据通常存储在不同的分区目录中，分区目录通常以“分区列名=值”的格式进行命名。
• 以people作为表名，gender和country作为分区列，给出存储数据的目录结构

（二）分区自动推断演示

1、建四个文件

• 在master虚拟机上/home里创建如下目录及文件，其中目录people代表表名，gender和country代表分区列，people.json存储实际人口数据

2、读取表数据

• 执行命令：spark-shell，启动Spark Shell
• 执行命令：val peopledf = spark.read.format("json").load("file:///home/people")

3、输出Schema信息

• 执行命令：peopledf.printSchema()

4、显示数据帧内容

• 执行命令：peopledf.show()
• 从输出的Schema信息和表数据可以看出，Spark SQL在读取数据时，自动推断出了两个分区列gender和country，并将这两列的值添加到了数据帧peopledf中。

（三）分区自动推断注意事项

• 分区列的数据类型是自动推断的，目前支持数字、日期、时间戳、字符串数据类型。若不希望自动推断分区列的数据类型，则可以在配置文件中将spark.sql.sources.partitionColumnTypeInference.enabled的值设置为false（默认为true，表示启用）。当禁用自动推断时，分区列将使用字符串数据类型。

到了这里，关于Spark大数据处理讲课笔记4.2 Spark SQL数据源 - 基本操作的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！