SparkSQL 读写_Reader | 学习笔记

开发者学堂课程【大数据 Spark 2020版（知识精讲与实战演练）第三阶段：SparkSQL 读写_Reader】学习笔记，与课程紧密联系，让用户快速学习知识。

课程地址：https://developer.aliyun.com/learning/course/690/detail/12052

SparkSQL 读写_Reader

 

内容介绍：

一、数据读取框架

二、初识 DataFrameReader

 

根据上章的讲述在 SparkSQL 中为了方便开发者，减少开发者的负担，为其设置了一套数据读取的新框架，本章主要介绍数据读写框架是什么，有哪些行为，如何使用和其中有什么内容的初步了解后再进行更深的讲解。

 

一、数据读取框架

大致分为三个步骤，首先看框架，大致有个位置。之后看数据读取框架中有什么内容，最后进行案例讲解来整合所有的知识点。

1.看看框架在哪

打开 IDEA,创建一个 Scala Class 叫做 ReadeWrite.

创建好以后再创建一个 reader 方法。编写任何程序之前都需要创建 SparkSession，之后再查看框架在哪。

l ReadWrite 代码如下：

package cn.itcast.spark.sql

import org.apache.spark.sql. Sparksession

import org.junit.Test

class Readwrite {

@Test

def reader1(): Unit = {

// 1．创建 Sparksession

//使用 SparkSession 的 bulider()，采用构建者模式进行相应创建

val spark = SparkSession.builder()

.master( master = "local[6]") //设置 master 为 local[6]

.appName( name= "reader1")//设置 appName 为 reader1

.getOrcreate() //最后通过一个 getOrCreate()创建 SparkSession

// 2．框架在哪，查看框架在哪，直接使用.read 即可，因为它就在 read 中

//将其赋给变量 reader,补全类型可以知道是 DataFrameReader 类型的。

//即每回在使用 read 时获取到的对象都是 DataFrameReader 类型的

val reader: DataFrameReader =spark.read

}

}

 

二、初识 DataFrameReader

1.DataFrameReader 内容

DataFrameReader 分为 Data 和 Reader 两个单词，其中 Reader 表示的是读取者，即读取数组的人，而 DataFrameReader 则表示读取出的数据是 DataFrame，所以就可以对其有个基本的认识，包括之后编写类名时看见类名就应该有个初步的认知。

l 理解 DataFrameReader 的整体结构和组成

SparkSQL 的一个非常重要的目标就是完善数据读取,所以 SparkSQL 中增加了一个新的框架,专门用于读取外部数据源,叫做 DataFrameReader

import org. apache.spark.sql.SparkSession

import org. apache.spark.sql.DataFrameReader

val spark:SparkSession= ...

val reader:DataFrameReader = spark.read

如果想要读取 csv 中的文件，至少需要设置的内容如下：

首先①要告知文件所在的位置，即文件地址。

②告知文件的类型是 csv 避免当成 json 读取，即文件类型。

③同时还有地址以及类型，而该怎么读取 mysql 中的表就需要设置一个 jdbc 的位置 url，所以还需要设置读取时候的参数。

④有了参数后 DataFrame 并不是一个放数组的容器，它代表了一个关系型数据库表不仅有表还有结构型数据，即结构信息。以上就是必须要有的四个点。

l DataFrameReader 由如下三大个组件组成，schema 就是结构

信息，option 就是连接外部的参数，而 format 就是格式。同时四要素的地址和文件也是会设置的。

 

 

2.小案例

DataFrameReader 有两种访问方式,一种是使用 load 方法加载,使用 format 指定加载格式,还有一种是使用封装方法,类似 csv, json , jdbc 等。

了解了 DataFrameReader 后，就进入小案例，该案例是由两种方法来去操作和读取 csv 文件的。从而让大家对 DataFrameReader 的使用有大概了解。

l 代码如下：

import org.apache.spark.sql. SparkSession

import org.apache.spark.sql. DataFrame

val spark: SparkSession = ...

//使用 load 方法

val fromLoad: DataFrame =spark

.read

.format("csv")

.option ("header" , true)

.option ( " inferSchema" , true)

.load("dataset/BeijingPM20100101_20151231.csv")

// Using format-specific load operator

val fromcSV: DataFrame = spark

. read

.option("header" , true)

.option ("inferSchema" , true)

.csv("dataset/BeijingPN20100101_20151231.csv")

打开 IDEA,在之前的类中创建一个新方法，第一步仍然是创建 SparkSession，但第二步是采用两种读写形式的，案例是读取 BeijingPM 的数据集为例。

l IDEA 代码案例如下：

/**

*初体验 Reader

*/

@Test

def reader2(): unit = {

//1．创建 SparkSession

val spark = Sparksession.builder()

.master( master= "local[6]")

.appName( name = "reader1")

.getorreate()

//2.第一种形式，也是最直观的形式，使用 spark.read，因为需要读取文件，所以设置文件的路径，同时不能仅仅依靠后缀名去猜测文件类型，所以还需要手动指定文件类型为 csv，以上就是 format 的设置，但此时打开 BeijingPM 的数据集文件后会发现第一行为 header,正常情况会将其当做一条数据而打印，所以就需要告知 DataFrameReader 不要将 header 打印,通过.option()进行配置设置为 true 类型。

最后因为 BeijingPM 还含有数据结构的类型，也需要打印，所以通过 option(“inferSchema”，value =true)进行一个数据的推断，其中 infer 为推断，Schema 为结构信息的含义，即此时的结构信息来自于 BeijingPM.csv 文件当中，最后 show()即可查看结果。

spark.read

.format( source= "csv")//设置文件格式

.option("header", value = true)//设置 header 不被读取

.option("inferschema", value = true)//设置读取结构型信息

.load( path = "dataset/BeijingPM20100101_20151231.csv")//设置文件路径

.show(numRow = 10)//查看 10 条结果

//3.第二种形式，是对于第一种形式的简写，可以直接通过 csv 属性对其地址进行读取，同时再给.option 设置使得不读取 header 的同时读取结构型信息。此时不需要在设置 format，因为.csv 处不仅指定了文件在哪，还指定了类型。

spark.read

.option("header", value = true)

.option( "inferschema", value = true)

.csv( path = "dataset/BeijingPw20100i01_2015123f.csv")//指定文件位置以及格式

.show()

}

以上就是两种形式，第一种是使用 load+format，第二种是具体使用文件类型的方法，将其运行就可以看到如下两种方式的查询结果：

（第一种方式）

 

（第二种方式）

得到以上结果说明操作过程没有问题。

3.csv 源码感悟

在进行总结的时候，先查看一下 csv 中的内容，点开 csv 进入其中如下，发现还是调用了一个 csv，再次按住 ctrl 点击进入。

l 进入后第一级代码

def csv(path: string): DataFrame = {

csv(seq(path):_*)

}

l 进入后第二级代码，可以看到 csv 也是调用了一个 format 方法，

同时还使用了一个 load，因而本质上，其实还是第一种形式，不过是为了一个新的格式去设置了一个支持即简写形式而已。

@scala.annotation.varargs

def csv(paths: string*): DataFrame =format( source ="csv"). load(paths : _*)

4.总结

①看看框架在哪

即框架该如何拿到，spark.read 即可。

②数据框架中有什么内容

包含路径，文件位置，schema 信息其次就是一些参数和类型。