Spark大数据分析与实战笔记（第四章 Spark SQL结构化数据文件处理-05） ...

逐日一句正能量

   努力学习，勤奋工作，让芳华更加光彩。
  第4章 Spark SQL结构化数据文件处理

章节概要

   在许多情况下，开发工程师并不了解Scala语言，也不了解Spark常用API，但又非常想要利用Spark框架提供的强盛的数据分析能力。Spark的开发工程师们考虑到了这个标题，利用SQL语言的语法简便、学习门槛低以及在编程语言遍及程度和流行程度高等诸多优势，从而开发了Spark SQL模块，通过Spark SQL，开发职员可以或许通过利用SQL语句，实现对结构化数据的处理。本章将针对Spark SQL的根本原理、利用方式举行详细解说。
  4.5 Spark SQL操作数据源

Spark SQL可以通过JDBC从关系型数据库中读取数据的方式创建DataFrame，通过对DataFrame举行一系列的操作后，还可以将数据重新写入到关系型数据中。关于Spark SQL对MySQL数据库的干系操作具体如下。
4.5.1 Spark SQL操作MySQL

• 读取MysQL数据库
  通过SQLyog工具远程毗连hadoop01节点的MySQL服务（这里选择的是SQLyog，用其它的工具也是一样的），并利用可视化操作界面创建名称为“spark"的数据库，并创建名称为“person”的数据表，以及向表中添加数据。
  同样也可以在hadoop01节点上利用MySQL客户端创建数据库、数据表以及插入数据，具体下令如下。
  

• 启动mysql客户端
  mysql -u root -p #屏幕提示输入暗码
  

结果如下图所示

• 创建spark数据库
  mysql > CREATE database spark ;
  

结果如下图所示

• 创建person数据表
  mysql > CREATE TABLE person (id INT(4) , NAME CHAR(20) , age INT(4));
  

结果如下图所示

• 插入数据
  

1. mysql > INSERT INTO person VALUE( 1 , ' zhangsan' , 18);
2. mysql > INSERT INTO person VALUE(2, ' lisi ' ,20);
3. mysql > SELECT * FROM person;

复制代码

结果如下图所示

数据库和数据表创建乐成后，如果想通过Spark SQL API方式访问MySQL数据库，必要在pom.xml设置文件中添加MySQL驱动毗连包，依赖参数如下。

1. <dependency>
2. <groupId>mysql</groupId>
3. <artifactId>mysql-connector-java</artifactId>
4. <version> 8.0.30 </version>
5. </ dependency>

复制代码

当所需依赖添加完毕后，就可以编写代码读取MySQL数据库中的数据，具体代码如文件所示：
SparkSqlToMysql.scala

1. package cn.itcast.sql
3. import java.util.Properties
5. import org.apache.spark.sql.{DataFrame, SparkSession}
7. //需要MySQL连接驱动包
8. object DataFromMysql {
9.     def main(args: Array[String]): Unit = {
10.         //1、创建sparkSession对象
11.         val spark: SparkSession = SparkSession.builder()
12.                 .appName("DataFromMysql")
13.                 .master("local[2]")
14.                 .getOrCreate()
15.         //2、创建Properties对象，设置连接mysql的用户名和密码
16.         val properties: Properties =new Properties()
17.         properties.setProperty("user","root")
18.         properties.setProperty("password","12345678")
19.         //读取mysql中的数据
20.         val mysqlDF : DataFrame = spark.read.jdbc("jdbc:mysql://192.168.121.128:3306/spark?useSSL=false","person",properties)
22.         var str = ""
23.         mysqlDF.collect().foreach(rdd => {
24.             str = str+rdd.get(0).toString+":"+rdd.get(1).toString
25.             val id = rdd
26.         })
27.         print(str)
29.         mysqlDF.show()
30.         spark.stop()
31.     }
32. }

复制代码

运行结果如下图所示

2. 向MySQL数据库写入数据
Spark SQL不仅可以或许查询MySQL数据库中的数据，还可以向表中插入新的数据，实现方式的具体代码如文件所示。
SparkSgIToMysql.scala

1. package cn.itcast.sql
3. import java.util.Properties
5. import org.apache.calcite.avatica.ColumnMetaData.StructType
6. import org.apache.spark.rdd.RDD
7. import org.apache.spark.sql.types.{IntegerType, StructField}
8. import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
10. object SparkSqlToMysql01 {
11.     def main(args: Array[String]): Unit = {
12.         //1.创建sparkSession对象
13.         val spark: SparkSession = SparkSession.builder()
14.                 .appName("SparkSqlToMysql")
15.                 .master("local[2]")
16.                 .getOrCreate()
17.         //2.读取数据
18.         //val data: RDD[String] = spark.sparkContext.textFile("D://spark//student.txt")
19.         val data: RDD[String] = spark.sparkContext.parallelize(Array("3,wangwu,22","4,zhaoliu,26"))
20.         //3.切分每一行,
21.         val arrRDD: RDD[Array[String]] = data.map(_.split(","))
22.         //4.RDD关联Student
23.         val personRDD: RDD[Student] = arrRDD.map(x=>Student(x(0).toInt,x(1),x(2).toInt))
24.         //导入隐式转换
25.         import spark.implicits._
26.         //5.将RDD转换成DataFrame
27.         val personDF: DataFrame = personRDD.toDF()
28.         //6.创建Properties对象，配置连接mysql的用户名和密码
29.         val prop =new Properties()
30.         prop.setProperty("user","root")
31.         prop.setProperty("password","12345678")
32.         prop.setProperty("driver","com.mysql.jdbc.Driver")
34.         personDF.write.mode("append").jdbc("jdbc:mysql://192.168.121.128:3306/spark?useUnicode=true&characterEncoding=utf8","spark.person",prop)
35.         personDF.show()
36.         spark.stop()
37.     }
38. }

复制代码

运行结果如下图所示

查察mysql中的数据表

4.5.2 操作Hive数据集

Apache Hive是Hadoop上的SQL引擎，也是大数据系统中紧张的数据仓库工具，Spark SQL支持访问Hive数据仓库，然后在Spark引擎中举行统计分析。接下来先容通过Spark SQL操作Hive数据仓库的具体实现步骤。

• 预备情况
  Hive采用MySQL数据库存放Hive元数据，因此为了可以或许让Spark访问Hive，就必要将MySQL驱动包拷贝到Spark安装路径下的jars目录下，具体下令如下。
  

1. cp mysql-connector-java-5.1.32.jar   /export/servers/spark/jars/

复制代码

结果如下图所示

要将Spark SQL毗连到一个摆设好的Hive时，就必须要把hive
-site.xml设置文件复制到Spark的设置文件目录中，这里采用软毗连方式，具体下令如下。
ln -s /export/servers/apache-hive
-1.2.1-bin/conf/hive
-site.xml \ /export/servers/spark/conf/hive
-site.xml
结果如下图所示

• 在Hive中创建数据库和表
  接下来，我们首先在hadoop01节点上启动Hive服务，创建数据库和表，具体下令如下所示。
  

启动hive
步伐

1. hive

复制代码

结果如下图所示

创建数据仓库

1. create database sparksqltest;

复制代码

结果如下图所示

创建数据表

1. create table if not exists sparksqltest.person(id int,name string,age int);

复制代码

结果如下图所示

切换数据库

1. use sparksqltest;

复制代码

结果如下图所示

向数据表中添加数据

1. insert into person values(1, "tom",29);
2. insert into person values(2, "jerry",20);

复制代码

结果如下图所示

目前，我们创建乐成了person数据表，并在该表中插入了两条数据，下面克隆hadoop01会话窗口，执行Spark-Shell。

• Spark SQL操作Hive数据库
  执行Spark-Shell，首先进入sparksqltest数据仓库，查察当前数据仓库中是否存在person表，具体代码如下所示。
  

1. spark-shell --master spark://hadoop01:7077
2. spark.sql("use sparksqltest")
3. spark.sql("show tables").show;

复制代码

结果如下图所示

如果spark没有启动的话，必要启动一下

show()方法背面的()因为没有参数，是可以去掉的

从上述返回结果看出，当前Spark-Shell乐成显示出Hive数据仓库中的person表。

• 向Hive表写入数据
  在插入数据之前，首先查察当前表中数据，具体代码如下所示。
  

1. spark.sql("select * from person").show

复制代码

从上述返回结果看出，当前person表中仅有两条数据信息。
下面在Spark-Shell中编写代码，添加两条数据到person表中，代码具体如下所示。

1. import java.util.Proerties
2. import org.apache.spark.sql.types._
3. import org.apache.spark.sql.Row
5. #创建数据
6. val personRDD = spark.sparkContest.parallelize(Array("3 zhangsan 22", "4 lisi 29")).map(_.split(" "))
8. #设置personRDD的Schema
9. val schema = StructType(List(StructField("id", IntegerType, true), StructField("name", StringType,true),StructField("age", IntegerType,true)))
11. #创建Row对象，每个Row对象都是rowRDD中的一行
12. val rowRDD = personRdd.map(p=>Row(p(0).toInt, p(1).trim, p(2).toInt))
14. #建立rowRDD与Schema对应关系，创建DataFrame
15. val personDF = spark.createDataFrame(rowRDD,schema)
17. #注册临时表
18. personDF.registerTempTable("t_person")
20. #将数据插入Hive表
21. spark.sql("insert into person select * from t_person")
23. #查询表数据
24. spark.sql("select * from person").show

复制代码

转载自：https://blog.csdn.net/u014727709/article/details/132515279
欢迎