Spark将Dataframe数据写入Hive分区表的方案_spark dataframe写入hive

《2021年最新版大数据面试题全面开启更新》

《2021年最新版大数据面试题全面开启更新》

 

DataFrame 将数据写入hive中时，默认的是hive默认数据库,insert into没有指定数据库的参数，数据写入hive表或者hive表分区中：

 

1、将DataFrame数据写入到hive表中

 

从DataFrame类中可以看到与hive表有关的写入API有一下几个：

registerTempTable(tableName:String):Unit,	
inserInto(tableName:String):Unit	
insertInto(tableName:String,overwrite:Boolean):Unit	
saveAsTable(tableName:String,source:String,mode:SaveMode,options:Map[String,String]):Unit

有很多重载函数，不一一列举

registerTempTable函数是创建spark临时表

insertInto函数是向表中写入数据，可以看出此函数不能指定数据库和分区等信息，不可以直接写入。

向hive数据仓库写入数据必须指定数据库，hive数据表建立可以在hive上建立，或者使用hiveContext.sql("create table .....")

下面语句是向指定数据库数据表中写入数据：

case class Person(name:String,col1:Int,col2:String)	
	
val sc = new org.apache.spark.SparkContext   	
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)	
	
import hiveContext.implicits._	
hiveContext.sql("use DataBaseName")	
val data = sc.textFile("path").map(x=>x.split("\\s+")).map(x=>Person(x(0),x(1).toInt,x(2)))	
data.toDF().insertInto("tableName")

 

创建一个case类将RDD中数据类型转为case类类型，然后通过toDF转换DataFrame,调用insertInto函数时，首先指定数据库，使用的是hiveContext.sql("use DataBaseName") 语句，就可以将DataFrame数据写入hive数据表中了。

 

2、将DataFrame数据写入hive指定数据表的分区中

 

hive数据表建立可以在hive上建立，或者使用hiveContext.sql("create table....") ,使用saveAsTable时数据存储格式有限，默认格式为parquet，将数据写入分区的思路是：首先将DataFrame数据写入临时表，之后由hiveContext.sql语句将数据写入hive分区表中，具体操作如下：

case class Person(name:String,col1:Int,col2:String)	
	
val sc = new org.apache.spark.SparkContext   	
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)	
	
import hiveContext.implicits._	
hiveContext.sql("use DataBaseName")	
val data = sc.textFile("path").map(x=>x.split("\\s+")).map(x=>Person(x(0),x(1).toInt,x(2)))	
data.toDF().registerTempTable("table1")	
hiveContext.sql("insert into table2 partition(date='2015-04-02') select name,col1,col2 from table1")

上面代码是spark版本1.6

 

下面代码是spark版本2.0及以上版本

val session = SparkSession.builder().appName("WarehouseInventoryByNewMysqlSnap").enableHiveSupport().getOrCreate()	
val sc: SparkContext=session.sparkContext	
session.sql("use bi_work")	
import session.implicits._	
val data = sc.textFile("path").map(x=>x.split("\\s+")).map(x=>Person(x(0),x(1).toInt,x(2)))	
data.toDF().registerTempTable("table1")	
session.sql("insert into table2 partition(date='2015-04-02') select name,col1,col2 from table1")

这边捎带介绍一下hive创建分区表：

hive分区表：是指在创建表时指定的partition的分区空间，若需要创建有分区的表，需要在create表的时候调用可选参数partitioned by。

注意：

• 一个表可以拥有一个或者多个分区，每个分区以文件夹的形式单独存在表文件夹的目录下

• hive的表和列名不区分大小写

• 分区是以字段的形式在表的结构中存在，通过desc table_name 命令可以查看到字段存在，该字段是分区的标识

• 建表的语句：

CREATE EXTERNAL TABLE bi_work.`dw_inventory_snap`	
(`warehouse_id` string COMMENT '',	
`internal_id` string COMMENT '', 	
`logical_inventory` string COMMENT '', 	
`create_time` timestamp COMMENT '')	
PARTITIONED BY (`snap_time` string) --指定分区	
row format delimited fields terminated by '\t'

 

— THE END —

◤半年文章精选系列◥

Flink从入门到放弃之源码解析系列

• 《Flink组件和逻辑计划》

• 《Flink执行计划生成》

• 《JobManager中的基本组件(1)》

• 《JobManager中的基本组件(2)》

• 《JobManager中的基本组件(3)》

• 《TaskManager》

• 《算子》

• 《网络》

• 《水印WaterMark》

• 《CheckPoint》

• 《任务调度及负载均衡》

• 《异常处理》

大数据成神之路-基础篇

• 《HashSet》

• 《HashMap》

• 《LinkedList》

• 《ArrayList/Vector》

• 《ConcurrentSkipListMap》

• 《ConcurrentHashMap1.7》

• 《ConcurrentHashMap1.8 Part1》

• 《ConcurrentHashMap1.8 Part2》

• 《CopyOnWriteArrayList》

• 《CopyOnWriteArraySet》

• 《ConcurrentLinkedQueue》

• 《LinkedBlockingDeque》

• 《LinkedBlockingQueue》

• 《ArrayBlockingQueue》

• 《ConcurrentSkipListSet》

大数据成神之路-进阶篇

• 《JVM&NIO基础入门》

• 《分布式理论基础和原理》

• 《分布式中的常见问题解决方案(分布式锁/事务/ID)》

• 《Zookeeper》

• 《RPC》

• 《Netty入门篇》

• 《Netty源码篇》

• 《Linux基础》

Flink入门系列

• 《Flink入门》

• 《Flink DataSet&DataSteam API》

• 《Flink集群部署》

• 《Flink重启策略》

• 《Flink分布式缓存》

• 《Flink广播变量》

• 《Flink中的Time》

• 《Flink中的窗口》

• 《时间戳和水印》

• 《Broadcast广播变量》

• 《Flink-Kafka-Connector》

• 《Flink之Table-&-SQL》

• 《Flink实战项目之实时热销排行》

• 《Flink-Redis-Sink》

• 《Flink消费Kafka写入Mysql》

Flink高级进阶

• 《FaultTolerance》

• 《流表对偶(duality)性》

• 《持续查询(ContinuousQueries)》

• 《DataStream-Connectors之Kafka》

• 《SQL概览》

• 《JOIN 算子》

• 《TableAPI》

• 《JOIN-LATERAL》

• 《JOIN-LATERAL-Time Interval(Time-windowed)》

• 《Temporal-Table-JOIN》

• 《State》

• 《FlinkSQL中的回退更新-Retraction》

• 《Apache Flink结合Apache Kafka实现端到端的一致性语义》

• 《Flink1.8.0发布！新功能抢先看》

• 《Flink1.8.0重大更新-Flink中State的自动清除详解》

• 《Flink在滴滴出行的应用与实践》

• 《批流统一计算引擎的动力源泉—Flink Shuffle机制的重构与优化》

• 《HBase分享 | Flink+HBase场景化解决方案》

• 《腾讯基于Flink的实时流计算平台演进之路》

• 《Flink进阶-Flink CEP(复杂事件处理)》

• 《Flink基于EventTime和WaterMark处理乱序事件和晚到的数据》

• 《Flink 最锋利的武器：Flink SQL 入门和实战》

• 《Flink Back Pressure》

• 《使用Flink读取Kafka中的消息》

• 《Flink on YARN部署快速入门指南》

• 《Apache Flink状态管理和容错机制介绍》

Hadoop生态圈系列

• 《Hadoop极简入门》

• 《MapReduce编程模型和计算框架架构原理》

• 《分布式文件系统-HDFS》

• 《YARN》

• 《Hadoop机架感知》

• 《HDFS的一个重要知识点-HDFS的数据流》

• 《Hadoop分布式缓存(DistributedCache)》

• 《如何从根源上解决 HDFS 小文件问题》（https://dwz.cn/FqDPpRUc）

• 《Hadoop解决小文件存储思路》（https://dwz.cn/2oCdmCkw）

• 《Hadoop所支持的几种压缩格式》

• 《MapReduce Join》

• 《YARN Capacity Scheduler（容量调度器）》

• 《hadoop上搭建hive》

• 《基于Hadoop的数据仓库Hive基础知识》

• 《Hive使用必知必会系列》

• 《一个小知识点-Hive行转列实现Pivot》

• 《面试必备技能-HiveSQL优化》

• 《HBase和Hive的区别和各自适用的场景》

• 《一篇文章入门Hbase》

• 《敲黑板：HBase的RowKey设计》

• 《HBase读写优化》

• 《HBase在滴滴出行的应用场景和最佳实践》

• 《Phoenix=HBase+SQL,让HBase插上了翅膀》

• 《一个知识点将你拒之门外之Hbase的二级索引》（https://dwz.cn/umfBOZ5l）

• 《Phoenix重磅 | Phoenix核心功能原理及应用场景介绍》

• 《DB、DW、DM、ODS、OLAP、OLTP和BI的概念理解》

• 《Hive/HiveSQL常用优化方法全面总结》

实时计算系列(spark、kafka等)

• 《Spark Streaming消费Kafka数据的两种方案》

• 《Apache Kafka简单入门》

• 《你不得不知道的知识-零拷贝》

• 《Kafka在字节跳动的实践和灾备方案》

• 《万字长文干货 | Kafka 事务性之幂等性实现》

• 《Kafka最佳实践》

• 《Kafka Exactly-Once 之事务性实现》

• 《Kafka连接器深度解读之错误处理和死信队列》

• 《Spark之数据倾斜调优》

• 《Structured Streaming 实现思路与实现概述》

• 《Spark内存调优》

• 《广告点击数实时统计：Spark StructuredStreaming + Redis Streams》

• 《Spark Shuffle在网易的优化》

• 《SparkSQL极简入门》

• 《下一代分布式消息队列Apache Pulsar》

• 《Pulsar与Kafka消费模型对比》

• 《Spark SQL重点知识总结》

• 《Structured Streaming 之状态存储解析》

• 《周期性清除Spark Streaming流状态的方法》

• 《Spark Structured Streaming特性介绍》

• 《Spark Streaming 反压（Back Pressure）机制介绍》

• 《Spark 从 Kafka 读数设置子并发度问题》

规范和系统设计

• 《阿里云10 PB+／天的日志系统设计和实现》

• 《阿里云Redis开发规范》

• 《Java中多个ifelse语句的替代设计》

• 《面试系列：十个海量数据处理方法大总结》

杂谈

• 《作为面试官的一点点感悟，谈谈技术人的成长之路》

• 《成年人的世界没有容易二字》

• 《我最近在关注的事》

• 《真香》

• 《简单说说学习这件事》

• 《20多岁做什么，将来才不会后悔》

• 《2019-05-12最近的总结》

• 《我军新闻联播气势+9999》

• 《周末分享 | 高手的战略》

• 《周末分享 | 快速定位自己的缺点》

• 《周末分享 | 我见过最高级的聪明是靠谱》