040 DataFrame中的write与read编程

一：SparkSQL支持的外部数据源

1.支持情况

2.External LIbraries

不是内嵌的，看起来不支持。

但是现在已经有很多开源插件，可以进行支持。

3.参考材料

·　　支持的格式：https://github.com/databricks

二：准备

1.启动服务

RunJar是metastore服务，在hive那边开启。

只需要启动三个服务就可以了，以后runjar都要启动，因为这里使用hive与spark集成了，不启动这个服务，就会总是报错。

2.启动spark-shell

三：测试检验程序

1.DataFrame的构成

2.结果

3.测试

4.结果

四：DataFrame的创建

1.创建SQLContext

val sqlContext=new SQLContext(sc)

2.创建DataFrame（两种方式）

val df=sqlContext.#

val df=sqlContext.read.#

3.DataFrame数据转换

val ndf=df.#.#

4.结果保存

ndf.#

ndf.write.#

五：DataFrame的保存

1.第一种方式

将DataFrame转换为RDD，RDD数据保存

2.第二种方式

直接通过DataFrame的write属性将数据写出。

但是有限制，必须有定义类实现，默认情况：SparkSQL只支持parquet，json，jdbc

六：两个常用的网站（数据源问题）

1.金砖公司提供的一些插件

2.package网址

https://spark-packages.org/

七：DataFrameReader编程模式

功能: 通过SQLContext提供的reader读取器读取外部数据源的数据，并形成DataFrame

1.源码的主要方法

format：给定数据源数据格式类型，eg: json、parquet
schema：给定读入数据的数据schema，可以不给定，不给定的情况下，进行数据类型推断
option：添加参数，这些参数在数据解析的时候可能会用到
load：
有参数的指从参数给定的path路径中加载数据，比如：JSON、Parquet...
无参数的指直接加载数据(根据option相关的参数)
jdbc：读取关系型数据库的数据
json：读取json格式数据
parquet：读取parquet格式数据
orc: 读取orc格式数据
table：直接读取关联的Hive数据库中的对应表数据

八：Reader的程序测试

1.新建文件夹

2.上传数据

3.加载json数据

val df=sqlContext.read.format("json").load("spark/sql/people.json")

结果：

4.数据展示

df.show()

结果：

5.数据注册成临时表并操作展示

结果：

6.和上面的方法等效的方式

sqlContext.sql("select * from json.`spark/sql/people.json`").show()

结果：

7.读取显示parquet格式的数据

sqlContext.read.format("parquet").load("spark/sql/users.parquet").show()

结果：

8.加载mysql中的数据

这个是服务器上的mysql。

sqlContext.read.jdbc("jdbc:mysql://linux-hadoop01.ibeifeng.com:3306/mysql?user=root&password=123456", "user", new java.util.Properties()).show()

这个地方比较特殊。

第一次使用bin/spark-shell进入后，使用命令，效果如下：

然后使用这种方式进行启动，加上jar

bin/spark-shell --jars /opt/softwares/mysql-connector-java-5.1.27-bin.jar --driver-class-path /opt/softwares/mysql-connector-java-5.1.27-bin.jar

九：DataFrameWriter编程模式

功能：将DataFrame的数据写出到外部数据源

1.源码主要方法

mode: 给定数据输出的模式
`overwrite`: overwrite the existing data.
`append`: append the data.
`ignore`: ignore the operation (i.e. no-op).
`error`: default option, throw an exception at runtime.
format：给定输出文件所属类型, eg: parquet、json
option: 给定参数
partitionBy：给定分区字段(要求输出的文件类型支持数据分区)
save: 触发数据保存操作 --> 当该API被调用后，数据已经写出到具体的数据保存位置了
jdbc：将数据输出到关系型数据库
当mode为append的时候，数据追加方式是：
先将表中的所有索引删除
再追加数据

没法实现，数据不存在就添加，存在就更新的需求

十：writer的程序测试

1.读取hive数据，形成DateFrame

2.结果保存为json格式

自动创建存储目录。

效果：

3.不再详细粘贴结果了

 读取Hive表数据形成DataFrame

 val df = sqlContext.read.table("common.emp")

 结果保存json格式

 df.select("empno","ename").write.mode("ignore").format("json").save("/beifeng/result/json")

 df.select("empno","ename").write.mode("error").format("json").save("/beifeng/result/json")

 df.select("empno","ename", "sal").write.mode("overwrite").format("json").save("/beifeng/result/json")

 df.select("empno","ename").write.mode("append").format("json").save("/beifeng/result/json")\

 上面虽然在追加的时候加上了sal，但是解析没有问题

 sqlContext.read.format("json").load("/beifeng/result/json").show()

 结果保存parquet格式

 df.select("empno", "ename", "deptno").write.format("parquet").save("/beifeng/result/parquet01")

 df.select("empno", "ename","sal", "deptno").write.mode("append").format("parquet").save("/beifeng/result/parquet01") ## 加上sal导致解析失败，读取数据的时候

 sqlContext.read.format("parquet").load("/beifeng/result/parquet01").show(100)

 sqlContext.read.format("parquet").load("/beifeng/result/parquet01/part*").show(100)

 partitionBy按照给定的字段进行分区

 df.select("empno", "ename", "deptno").write.format("parquet").partitionBy("deptno").save("/beifeng/result/parquet02")

 sqlContext.read.format("parquet").load("/beifeng/result/parquet02").show(100)