更新时间:2021年04月14日16时43分 来源:传智教育 浏览次数:
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表中,代码具体如下所示。
scala > import java.util.Properties scala > import org.apache.spark.sql.types._ scala > import org.apache.spark.sql.Row #创建数据 scala > val personRDD = spark.sparkContext .parallelize(Array("3 zhangsan 22","4 lisi 29")).map(_.split(" ")) #设置personRDD的Schema scala > val schema = StructType(List( StructField("id",IntegerType,true), StructField("name",StringType,true), StructField("age",IntegerType,true))) #创建Row对象,每个Row对象都是rowRDD中的一行 scala > val rowRDD = personRDD.map(p => Row(p(0).toInt,p(1).trim,p(2).toInt)) #建立rowRDD与Schema对应关系,创建DataFrame scala > val personDF = spark.createDataFrame(rowRDD,schema) #注册临时表 scala > personDF.registerTempTable("t_person") #将数据插入Hive表 scala > spark.sql("insert into person select * from t_person") #查询表数据 scala > spark.sql("select * from person").show +---+--------+---+ | id| name|age| +---+--------+---+ | 1| tom| 29| | 2| jerry| 20| | 3|zhangsan| 22| | 4| lisi | 29| +---+--------+---+
上述代码中,第5-6行代码表示先创建2条数据,并将其转换为RDD格式,由于Hive表中含有Schema信息,因此我们在第8-12行代码中采用编程方式定义Schema信息,第14-17行代码表示创建相应的DataFrame对象,第19-23行代码表示通过DataFrame对象向Hive表中插入新数据,从24-31行代码看出,数据已经成功插入到Hive表中。
猜你喜欢: