操作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
#创建数据仓库
hive > create database sparksqltest;
#创建数据表
hive > create table if not exists \
sparksqltest.person(id int,name string,age int);
#切换数据库
hive > use sparksqltest;
#向数据表中添加数据
hive > insert into person values(1,"tom",29);
hive > 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
scala > spark.sql("use sparksqltest")
res0: org.apache.spark.sql.DataFrame = []
scala > spark.sql("show tables").show;
+------------+---------+-----------+
| database  |tableName|isTemporary|
+------------+---------+-----------+
|sparksqltest| person |  false  |
+------------+---------+-----------+

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

4．向Hive表写入数据

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

scala> spark.sql("select * from person").show
+---+--------+---+
| id| name |age|
+---+--------+---+
| 1|  tom  | 29|
| 2| jerry  | 20|
+---+--------+---+

从上述返回结果看出，当前person表中仅有两条数据信息。

下面在Spark-Shell中编写代码，添加两条数据到person表中，代码具体如下所示。

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

上述代码中，第5-6行代码表示先创建2条数据，并将其转换为RDD格式，由于Hive表中含有Schema信息，因此我们在第8-12行代码中采用编程方式定义Schema信息，第14-17行代码表示创建相应的DataFrame对象，第19-23行代码表示通过DataFrame对象向Hive表中插入新数据，从24-31行代码看出，数据已经成功插入到Hive表中。