上海正规网站建设耗材aso平台

每日一句正能量

努力学习，勤奋工作，让青春更加光彩。

第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

1. 读取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));

结果如下图所示

• 插入数据

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

结果如下图所示

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

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

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

package cn.itcast.sqlimport java.util.Propertiesimport org.apache.spark.sql.{DataFrame, SparkSession}//需要MySQL连接驱动包
object DataFromMysql {def main(args: Array[String]): Unit = {//1、创建sparkSession对象val spark: SparkSession = SparkSession.builder().appName("DataFromMysql").master("local[2]").getOrCreate()//2、创建Properties对象，设置连接mysql的用户名和密码val properties: Properties =new Properties()properties.setProperty("user","root")properties.setProperty("password","12345678")//读取mysql中的数据val mysqlDF : DataFrame = spark.read.jdbc("jdbc:mysql://192.168.121.128:3306/spark?useSSL=false","person",properties)var str = ""mysqlDF.collect().foreach(rdd => {str = str+rdd.get(0).toString+":"+rdd.get(1).toStringval id = rdd})print(str)mysqlDF.show()spark.stop()}
}

运行结果如下图所示

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

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

运行结果如下图所示

查看mysql中的数据表

4.5.2 操作Hive数据集

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

1. 准备环境
   Hive采用MySQL数据库存放Hive元数据，因此为了能够让Spark访问Hive，就需要将MySQL驱动包拷贝到Spark安装路径下的jars目录下，具体命令如下。

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
结果如下图所示

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

启动hive程序

hive

结果如下图所示

创建数据仓库

create database sparksqltest;

结果如下图所示

创建数据表

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

结果如下图所示

切换数据库

use sparksqltest;

结果如下图所示

向数据表中添加数据

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

结果如下图所示

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

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

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

结果如下图所示

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


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

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

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

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

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

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

转载自：https://blog.csdn.net/u014727709/article/details/132515279
欢迎 👍点赞✍评论⭐收藏，欢迎指正