博客

数据的加载和保存

1通用的加载和保存方式

SparkSQL 提供了通用的保存数据和数据加载的方式。

这里的通用指的是

使用相同的API，根据不同的参数读取和保存不同格式的数据

，

SparkSQL 默认读取和保存的文件格式为 parquet

1)加载数据


spark.read.load 是加载数据的通用方法


scala> spark.read.
csv table    format text    jdbc    json textFile    load    option    options    orc    parquet    schema

如果读取不同格式的数据，可以对不同的数据格式进行设定

scala> spark.read.format("…")[.option("…")].load("…")

• format(“…”)：指定加载的数据类型，包括”csv”、“jdbc”、“json”、“orc”、“parquet”和”textFile”。
• load(“…”)：在”csv”、“jdbc”、“json”、“orc”、”parquet”和”textFile”格式下需要传入加载数据的路径。
• option(“…”)：在”jdbc”格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable 我们前面都是使用read API 先把文件加载到 DataFrame 然后再查询

上传

users.parquet

到

/opt/module/spark-standalone


val df = spark.read.format("parquet").load("users.parquet")
df.show()
df.write.save("output")
//save指定目录

其实，我们也可以直接在文件上进行查询: 文件格式.`文件路径`

上传文件

employees.json

到

/opt/module/spark-standalone


执行以下代码

scala>spark.sql("select * from json.`/opt/module/spark-standalone/employees.json`").show

2)保存数据

df.write.save 是保存数据的通用方法

如果保存不同格式的数据，可以对不同的数据格式进行设定

scala>df.write.format("…")[.option("…")].save("…")    

• format(“…”)：指定保存的数据类型，包括”csv”、“jdbc”、“json”、“orc”、“parquet”和”textFile”。
• save (“…”)：在”csv”、“orc”、”parquet”和”textFile”格式下需要传入保存数据的路径。
• option(“…”)：在”jdbc”格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable
  
  保存操作可以使用 SaveMode, 用来指明如何处理数据，使用 mode()方法来设置。
  
  有一点很重要: 这些 SaveMode 都是没有加锁的, 也不是原子操作。
  
  SaveMode 是一个枚举类，其中的常量包括：

df.write.mode("append").json("/opt/module/data/output")

2Parquet

Spark SQL 的默认数据源为 Parquet 格式。

Parquet 是一种能够有效存储嵌套数据的

列式存储格式

。

数据源为 Parquet 文件时，Spark SQL 可以方便的执行所有的操作，不需要使用 format。

修改配置项spark.sql.sources.default，可修改默认数据源格式。

1)加载数据

scala> val df = spark.read.load("/opt/module/spark-standalone/users.parquet")
scala> df.show

2)保存数据

scala> var df = spark.read.json("/opt/module/spark-standalone/employees.json")
//保存为 parquet 格式
scala> df.write.mode("append").save("/opt/module/spark-standalone/output2")

3JSON

Spark SQL 能够自动推测 JSON 数据集的结构，并将它加载为一个Dataset[Row].

可以通过 SparkSession.read.json()去加载 JSON 文件。

注意：Spark 读取的 JSON 文件不是传统的JSON 文件，每一行都应该是一个 JSON 串。格式如下：

{"name":"Michael"}
{"name":"Andy","age":30}
[{"name":"Justin1","age":19},{"name":"Justin2","age":19}]

1）导入隐式转换

import spark.implicits._

2）加载 JSON 文件

val path = "/opt/module/spark-standalone/people.json"
val peopleDF = spark.read.json(path)

3）创建临时表

peopleDF.createOrReplaceTempView("people")

4）数据查询

val teenagerNamesDF = spark.sql("SELECT    name    FROM    people    WHERE    age    BETWEEN    13 AND 19")
teenagerNamesDF.show()

±—–+

| name|

±—–+

|Justin|

±—–+

4 CSV

Spark SQL 可以配置 CSV 文件的列表信息，读取CSV 文件,CSV 文件的第一行设置为数据列

spark.read.format("csv").option("sep", ";").option("inferSchema","true").option("header", "true").load("people.csv")

infer    英[ɪnˈfɜː(r)] 美[ɪnˈfɜːr] v.    推断; 推论; 推理; 间接地提出; 暗示; 意指;

5MySQL

Spark SQL 可以通过 JDBC 从关系型数据库中读取数据的方式创建DataFrame，

通过对DataFrame 一系列的计算后，还可以将数据再写回关系型数据库中。

如果使用 spark-shell 操作，可在启动shell 时指定相关的数据库驱动路径或者将相关的数据库驱动放到 spark 的类路径下。

上传驱动包到

/opt/module/spark-standalone


bin/spark-shell  --jars mysql-connector-java-5.1.27-bin.jar

5.1 spark shell 读 MySQL

准备数据库

create database spark_sql;
use spark_sql;
create table `user`(
	name varchar(20),
	age int
)
insert into `user` values('jack',20);
insert into `user` values('rose',20);
insert into `user` values('tony',21);
delete from `user`;

在spark-shell命令行执行

spark.read.format("jdbc")
.option("url", "jdbc:mysql://hadoop102:3306/spark_sql")
.option("driver", "com.mysql.jdbc.Driver")
.option("user", "root")
.option("password", "000000")
.option("dbtable", "user")
.load()
.show

5.2 spark shell 写 MySQL

case class User2(name: String, age: Long)


val rdd=spark.sparkContext.makeRDD(List(User2("jack", 10),User2("rose", 10),User2("tony", 20)))
 
val ds = rdd.toDS

ds.write
.format("jdbc")
.option("url", "jdbc:mysql://hadoop102:3306/spark_sql")
.option("driver", "com.mysql.jdbc.Driver")
.option("user", "root")
.option("password", "000000")
.option("dbtable", "user")
.mode("append")
.save

5.3 idea编写spark sql 读写 MySQL

1）导入依赖

<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.27</version>
</dependency>

2）读取数据

val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
//创建 SparkSession 对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate() import spark.implicits._
//方式 1：通用的 load 方法读取
spark.read.format("jdbc")
.option("url", "jdbc:mysql://hadoop102:3306/spark_sql")
.option("driver", "com.mysql.jdbc.Driver")
.option("user", "root")
.option("password", "000000")
.option("dbtable", "user")
.load().show

//方式 2:通用的 load 方法读取 参数另一种形式
spark.read.format("jdbc")
.options(Map("url"->"jdbc:mysql://hadoop102:3306/spark_sql?user=root&password=000000",
"dbtable"->"user","driver"->"com.mysql.jdbc.Driver")).load().show

//方式 3:使用 jdbc 方法读取
val props: Properties = new Properties() props.setProperty("user", "root") props.setProperty("password", "000000")
val df: DataFrame = spark.read.jdbc("jdbc:mysql://hadoop102:3306/spark_sql", "user", props)
df.show

//释放资源
spark.stop()

3）写入数据

case class User2(name: String, age: Long)
。。。
val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")

//创建 SparkSession 对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate() import spark.implicits._

val rdd: RDD[User2] = spark.sparkContext.makeRDD(List(User2("lisi", 20), User2("zs", 30)))
val ds: Dataset[User2] = rdd.toDS
//方式 1：通用的方式 format 指定写出类型
ds.write
.format("jdbc")
.option("url", "jdbc:mysql://hadoop102:3306/spark_sql")
.option("user", "root")
.option("password", "000000")
.option("dbtable", "user")
.mode(SaveMode.Append)
.save()

//方式 2：通过 jdbc 方法
val props: Properties = new Properties()
props.setProperty("user", "root") 
props.setProperty("password", "000000")
ds.write.mode(SaveMode.Append).jdbc("jdbc:mysql://hadoop102:3306/spark_sql", "user", props)

//释放资源spark.stop()

6 Hive

Apache Hive 是 Hadoop 上的 SQL 引擎，Spark SQL 编译时可以包含 Hive 支持，也可以不包含。

包含 Hive 支持的 Spark SQL 可以支持 Hive 表访问、UDF (用户自定义函数)以及 Hive 查询语言(HiveQL/HQL)等。

需要强调的一点是，如果要在 Spark SQL 中包含Hive 的库，并不需要事先安装 Hive。一般来说，最好还是在编译 Spark SQL 时引入 Hive 支持，这样就可以使用这些特性了。

如果你下载的是二进制版本的 Spark，它应该已经在编译时添加了 Hive 支持。

若要把 Spark SQL 连接到一个部署好的 Hive 上，你必须把 hive-site.xml 复制到Spark 的配置文件目录中($SPARK_HOME/conf)。即使没有部署好 Hive，Spark SQL 也可以运行。 需要注意的是，如果你没有部署好 Hive，Spark SQL 会在当前的工作目录中创建出自己的 Hive 元数据仓库，叫作 metastore_db。此外，如果你尝试使用 HiveQL 中的CREATE TABLE (并非 CREATE EXTERNAL TABLE)语句来创建表，这些表会被放在你默认的文件系统中的 /user/hive/warehouse 目录中(如果你的 classpath 中有配好的

hdfs-site.xml，默认的文件系统就是 HDFS，否则就是本地文件系统)。

spark-shell 默认是Hive 支持的；代码中是默认不支持的，需要手动指定（加一个参数即可）。

1）内嵌的 HIVE （了解）

如果使用 Spark 内嵌的 Hive, 则什么都不用做, 直接使用即可.

scala> spark.sql("show tables").show
。。。
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
+--------+---------+-----------+

scala> spark.sql("create table aa(id int)")

。。。

scala> spark.sql("show tables").show
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
| default|    aa|    false|
+--------+---------+-----------+

Hive 的元数据存储在 derby 中, 默认仓库地址:$SPARK_HOME/spark-warehouse

向表加载本地数据

scala> spark.sql("load data local inpath 'input/ids.txt' into table aa")

。。。

scala> spark.sql("select * from aa").show
+---+
| id|
+---+
| 1|
| 2|
| 3|
+---+

在实际使用中, 几乎没有任何人会使用内置的 Hive

2）外部的 HIVE

如果想连接外部已经部署好的Hive，需要通过以下几个步骤：

Spark 要接管 Hive 需要把hive-site.xml 拷贝到conf/目录下

把 Mysql 的驱动 copy 到 jars/目录下

如果访问不到 hdfs，则需要把 core-site.xml 和 hdfs-site.xml 拷贝到 conf/目录下

cd /opt/module
find ./ -name hive-site.xml
scp ./conf/hive-site.xml  /opt/module/spark-standalone/conf/
find ./ -name hdfs-site.xml
scp ./hadoop-2.7.2/etc/hadoop/core-site.xml  /opt/module/spark-standalone/conf/
find ./ -name hdfs-site.xml
scp ./hadoop-2.7.2/etc/hadoop/hdfs-site.xml  /opt/module/spark-standalone/conf/

重启 spark-shell

bin/spark-shell

spark.sql("show tables").show

MetaException(message:Hive Schema version 2.3.0 does not match metastore’s schema version 1.2.0的错误。

更换hive的2.3.0版本

apache-hive-2.3.0-bin.tar.gz


多了一步：初始化原数据

schematool -dbType mysql -initSchema