Spark SQL 操作 MySQL数据库和 Hive数据仓库

r囧r小猫 2022-11-19 06:25 645阅读 0赞

### Spark SQ操作 MySQL和 Hive ###

*   *  一、启动各组件
     *  二、操作 MySQL
     *   *  2.1 读取 MySQL数据库
         *  2.2 Spark SQL API 操作 MySQL数据库
         *  2.3 Spark SQL 查询 MySQL数据
         *  2.4 向MySQL 写入 数据
     *  三、操作 Hive数据集
     *   *  3.1 准备环境配置
         *  3.2 在 Hive中创建数据库和表
         *  3.3 Spark SQL 操作 Hive数据库
         *  3.4 向 Hive写入数据

--------------------

Mysql+Hive：[1、Centos7 MySQL安装 —— 用网盘简单安装][1_Centos7 MySQL_ _]

[2、Hadoop集群搭建及配置⑨——Hive 可靠的安装配置][2_Hadoop_Hive]

[3、Spark SQ操作 MySQL数据库和 Hive数据仓库][3_Spark SQ_ MySQL_ Hive]

[4、Spark SQL RDD基本操作、RDD—DataFrame、API MySQL][4_Spark SQL RDD_RDD_DataFrame_API MySQL]

[5、Spark SQL RDD、DataFrame、Dataset、反射推断机制 Schema 操作！！][4_Spark SQL RDD_RDD_DataFrame_API MySQL]

--------------------

上次以及安装好了 MySQL和与Hive，这次我们就对MySQL与Hive进行一些操作。  
还没有安装的可以点击连接进行安装哟。

这次我们用Spark SQL 通过 DataFrame、Dataset操作多种数据源执行SQL查询，并且提供了多种数据源之间的转换方式，接下来，我们一起动手试试吧！！

--------------------

[8、Hadoop集群搭建及配置⑥ —— Hadoop组件安装及配置][8_Hadoop_ _ Hadoop]

[9、Hadoop集群搭建及配置⑦—— Spark&Scala安装配置][9_Hadoop_ Spark_Scala]

[10、Hadoop集群搭建及配置⑧——Hbase的安装配置][10_Hadoop_Hbase]

[11、eclipse配置连接Hadoop][11_eclipse_Hadoop]

[12、eclipse 实现 Hdfs java API][12_eclipse _ Hdfs java API]

[13、eclipse 实现 HBase java API][13_eclipse _ HBase java API]

[14、Hbase java API 实现增删改查][14_Hbase java API]

--------------------

## 一、启动各组件 ##

1.  启动 zookeeper
2.  启动 hadoop
3.  启动 spark
4.  MySQL是默认开机自启的
5.  slave1启动Hive服务端
6.  master 作为客户端开启 hive

cd /usr/zookeeper/zookeeper-3.4.10（此目录下）
    # 1、三个节点都启动 zookeeper
    bin/zkServer.sh start
    bin/zkServer.sh status （注意已经生效环境变量，环境变量开机重启后要手动再生效！！）
    bin/zkServer.sh stop（关闭）
    
    # 2、master 节点启动 hadoop
    /usr/hadoop/hadoop-2.7.3/sbin/start-all.sh
    
    # 3、master 节点启动 spark
    /usr/spark/spark-2.4.0-bin-hadoop2.7/sbin/start-all.sh
    
    # 4、slave 节点开启
    /usr/spark/spark-2.4.0-bin-hadoop2.7/sbin/start-master.sh
    
    # ~ 进入Scala 交互式，输入：~
    spark-shell --master local[2]
    
    # ~ 进入MySQL，输入：~
    mysql -uroot -pabc123456
    
    # 5、slave1启动 Hive服务端（先）
    /usr/hive/hive-2.1.1/bin/hive --service metastore
    
    # 6、master 作为客户端开启 hive（后）
    /usr/hive/hive-2.1.1/bin/hive

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 3]

*  启动完毕，接下来进行操作。

--------------------

## 二、操作 MySQL ##

### 2.1 读取 MySQL数据库 ###

# 0.启动mysql数据库
    mysql -uroot -pabc123456
    # 查看数据库
    show databases;
    # 2.创建数据库 test
    create database test;
    # 3.进入 test 数据库
    use test;
    # 4.创建 person 数据表
    create table person(id int(4),name char(20), age int(4));
    # 5.插入数据
    insert into person values(1,"lubu",30);
    insert into person values(2,"guanyu",32);
    # 6.查询person数据
    select * from person;

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 4]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 5]

### 2.2 Spark SQL API 操作 MySQL数据库 ###

数据库和数据表创建成功后，通过Spark SQL API 访问 MySQL数据库，需要再pom.xml 配置文件添加 MySQL驱动包。

*  【注】：在 安装Hive以及下载过了，可以略过这一步！！

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 6]  
**没有就添加配置：**

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

--------------------

添加完MySQL驱动包，编写代码对Mysql进行操作

### 2.3 Spark SQL 查询 MySQL数据 ###

import java.util.Properties
    import org.apache.spark.sql.{
        DataFrame, SparkSession}
    
    object DataFrameMysql {
        
      def main(args: Array[String]): Unit = {
        
        // 1、创建 SparkSession对象
        val spark:SparkSession = SparkSession.builder()
          .appName("DataFrameMysql")
          .master("local[2]")
          .getOrCreate()
    
        // 2.创建 Properties对象，设置连接MySQL的用户名和密码
        val properties:Properties = new Properties() //
        properties.setProperty("user","root")
        properties.setProperty("password","abc123456")
    
        // 3.读取MySQL的数据
        val mysqlDF : DataFrame = spark.read
          .jdbc("jdbc:mysql://master:3306/test", "person", properties)
    
        /*
        spark.read.jdbc 实现对MySQL的操作，三个参数分别是：
        JDBC的url、数据表名、数据库的用户名和密码
        ！！ 很容易写错，注意 ！！
         */
    
        // 4.显示MySQL表的数据
        mysqlDF.show()
        spark.stop()
      }
    }

--------------------

### 2.4 向MySQL 写入 数据 ###

import java.util.Properties
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{
        DataFrame, SparkSession}
    
    // 0. 创建 person样例类
    case class person(id:Int,name:String,age:Int)
    object SparkToMysql {
        
      def main(args: Array[String]): Unit = {
        
        // 1、创建 SparkSession对象
        val spark:SparkSession = SparkSession.builder()
          .appName("SparkToMysql")
          .master("local[2]")
          .getOrCreate()
    
        /*
        * spark.sparkContext.parallelize()方法创建一个RDD，RDD有 person两个数据；
        * 用","逗号 作为样例类字段匹配的依据，且转换为DataFrame对象；
        */
        // 2、创建插入数据
        val data = spark.sparkContext.parallelize(Array("3,lisi,19","4,xiaohong,15"))
        // 3、按列名切分
        val arrRDD:RDD[Array[String]] = data.map(_.split(","))
        // 4、RDD关联 person类
        val personRDD:RDD[person] = arrRDD.map(x=>person(x(0).toInt,x(1),x(2).toInt))
    
        // 导入隐式转换
        import spark.implicits._
        // 5、将 RDD转换成 DataFrame
        val personDF: DataFrame = personRDD.toDF()
        // 6.创建 Properties对象，设置连接MySQL的用户名和密码
        val pro:Properties = new Properties()
        pro.setProperty("user","root")
        pro.setProperty("password","abc123456")
        pro.setProperty("driver","com.mysql.jdbc.Driver")
        // 7、写入数据
        /*
        * personDF.write.mode() 方法表示设置写入数据方式，
        * append表示追加数据，overwrite表示覆盖数据；
        * errorIfExists表示表存在就报错，ignore 表示忽略新保存的数据.
        * */
        personDF.write.mode("append").jdbc("jdbc:mysql://master:3306/test", "test.person", pro)
        personDF.show()
        spark.stop()
      }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 7]

--------------------

## 三、操作 Hive数据集 ##

### 3.1 准备环境配置 ###

Hive 采用 MySQL数据库存放 Hive元数据，因此为了能够让 Spark访问 Hive，需要将 MySQL驱动包复制到 Spark安装路径下的jars目录下。

master和slave1 都要把 `mysql-connector-java-5.1.46.jar` 驱动包 放在 `/usr/spark/spark-2.4.0-bin-hadoop2.7/jars/` 目录下，

*  可以用 安装Hive时候 的IDEA方式把驱动器拖到spark的jars目录下。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 8]

也可以用命令 复制到spark的jars目录下。

# 复制本节点（salve1）
    cp /usr/hive/hive-2.1.1/lib/mysql-connector-java-5.1.46.jar /usr/spark/spark-2.4.0-bin-hadoop2.7/jars
    
    # 远程复制到master的spark的jars目录下，过程需要输入ssh密码：123456 (登录密码)
    scp -r /usr/hive/hive-2.1.1/lib/mysql-connector-java-5.1.46.jar root@master: /usr/spark/spark-2.4.0-bin-hadoop2.7/jars

![在这里插入图片描述][20210414215110452.png]  
![在这里插入图片描述][20210414215914756.png]  
要把Spark SQL 连接到一个部署好的Hive时，必须把 `hive-site.xml` 配置文件复制到 Spark的配置目录 conf上，采用软连接。

# 只在master操作
    ln -s /usr/hive/hive-2.1.1/conf/hive-site.xml /usr/spark/spark-2.4.0-bin-hadoop2.7/conf/hive-site.xml

![在这里插入图片描述][20210414220958406.png]

--------------------

### 3.2 在 Hive中创建数据库和表 ###

已经开启 Hive了。

# 5、slave1启动 Hive服务端（先）
    /usr/hive/hive-2.1.1/bin/hive --service metastore
    
    # 6、master 作为客户端开启 hive（后）
    /usr/hive/hive-2.1.1/bin/hive

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 9]

# 查看数据库
    show databases;
    # 1、创建数据仓库 sptest
    create database sptest;
    # 2.进入 sptest 数据库
    use sptest;
    # 3.创建数据表 person
    create table if not exists
    sptest.person(id int,name string,age int);
    
    # 4.插入数据
    insert into person values(1,"lubu",30);
    insert into person values(2,"guanyu",32);

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 10]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 11]  
![在这里插入图片描述][20210414222853984.png]

在 hive 交互式操作插入数据会很慢，所以我们用Spark SQL 进行操作会快得多。

用shell 再连接一个 master，进入spark-shell 交互式  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 12]

--------------------

### 3.3 Spark SQL 操作 Hive数据库 ###

# 1.进入 sptest数据库
    spark.sql("use sptest")
    # 2.显示数据表状态
    spark.sql("show tables").show;
    # 3.显示数据表数据
    spark.sql("select * from person").show

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 13]

--------------------

### 3.4 向 Hive写入数据 ###

import java.util.Properties;
    import org.apache.spark.sql.types._
    import org.apache.spark.sql.Row
    # 1、创建数据
    val personRDD = spark.sparkContext.parallelize(Array("3,zhangsan,22","4,lisi,19")).map(_.split(","))
    val schema = StructType(List(
    StructField("id",IntegerType,true),
    StructField("name",StringType,true),
    StructField("age",IntegerType,true)))
    # 2、创建rowRdd对象
    val rowRDD = personRDD.map(x=>Row(x(0).toInt,x(1).trim,x(2).toInt))
    # 3、建立 rowRDD 与 schema 对应关系，创建 DataFrame
    val personDF =spark.createDataFrame(rowRDD,schema)
    # 4、注册临时表
    personDF.registerTempTable("t_person")
    # 5、将数据插入Hive表
    spark.sql("insert into person select * from t_person")
    # 6、查询表数据
    spark.sql("select * from person").show

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 14]  
可以看到已经插入数据了，感觉可还行？？

[1_Centos7 MySQL_ _]: https://blog.csdn.net/weixin_44775255/article/details/115662187
[2_Hadoop_Hive]: https://blog.csdn.net/weixin_44775255/article/details/115663334
[3_Spark SQ_ MySQL_ Hive]: https://blog.csdn.net/weixin_44775255/article/details/115707963
[4_Spark SQL RDD_RDD_DataFrame_API MySQL]: https://blog.csdn.net/weixin_44775255/article/details/115727069
[8_Hadoop_ _ Hadoop]: https://blog.csdn.net/weixin_44775255/article/details/113775449
[9_Hadoop_ Spark_Scala]: https://blog.csdn.net/weixin_44775255/article/details/114518207
[10_Hadoop_Hbase]: https://blog.csdn.net/weixin_44775255/article/details/114676669
[11_eclipse_Hadoop]: https://blog.csdn.net/weixin_44775255/article/details/114799064
[12_eclipse _ Hdfs java API]: https://blog.csdn.net/weixin_44775255/article/details/114809298
[13_eclipse _ HBase java API]: https://blog.csdn.net/weixin_44775255/article/details/114821297
[14_Hbase java API]: https://blog.csdn.net/weixin_44775255/article/details/117446315
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70]: /images/20221022/452d24e0d3e14eb7b7865c55b377c396.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 1]: /images/20221022/d1e99ae6867f443b91c7eeb8e6474889.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 2]: /images/20221022/0db94d9854554897b838df69144a1a27.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 3]: /images/20221022/3acf7629a82d48dd8dfd47549dff6e0e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20210414205629646.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 5]: /images/20221022/7b850b75eb5540898c75547d292be567.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 6]: /images/20221022/3748d6bc293f4dd08b3d815aa40d0e2c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 7]: /images/20221022/cd1f6922c285450d9176f8077da1831a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 8]: /images/20221022/3f3820a6f59b4d9f90d15d9078616117.png
[20210414215110452.png]: /images/20221022/5ff005e701b04e9d9e6f21ce1c2db4af.png
[20210414215914756.png]: /images/20221022/93788b34f54b4be48af1858fad8e2324.png
[20210414220958406.png]: /images/20221022/b99d8c95729c4cfe8f34a6862233a134.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 9]: /images/20221022/4d8d1a83b5934b688aa530d5f7c50c99.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 10]: /images/20221022/2655518880e4480c89ad53caf204a8c6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 11]: /images/20221022/e017e00826a241679226ef909cebb948.png
[20210414222853984.png]: /images/20221022/cddc095b12694da4b3e0fed429c27113.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 12]: /images/20221022/784a8715e1a24d8d94cde894c0d50339.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 13]: /images/20221022/8a3850e4e20c4af5b81599afd38bf05b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDc3NTI1NQ_size_16_color_FFFFFF_t_70 14]: /images/20221022/64591aa4b4494300b56e431ba61652af.png