SparkSQL overwrite插入Hive表数据重复问题

问题描述：

使用Spark SQL采用overwrite写法写入Hive（非分区表，），全量覆盖，因为人为原因脚本定时设置重复，SparkSql计算任务被短时间内调起两次，结果发现任务正常运行，造成写入表中数据结果存在同一张表有重复的行，数据翻倍。

从hdfs上可以看到也存在重复的的数据文件，会。有两组文件，每组大小是一样的。

hdfs dfs -ls /user/hive/warehouse/xxx.db/xxx_table

问题思考：

如果存在多个任务同时往一张Hive表overwrite，因为资源等因素，也会有时间差，基本不可能两个任务同时同秒执行结束，执行的写入sql都是insert overwrite，因此数据也理论上是正常写入数据之前要删除旧的数据，覆盖才合理。猜想是可能Hive本身会有延迟，在短时间内上一个任务还未insert overwrite结束，另外一个任务也紧跟着运行insert overwrite结束，导致重复插入数据没有被覆盖。

经过查资料发现：

Spark SQL在执行SQL的overwrite的时候并没有删除旧的的数据文件（Spark SQL生成的数据文件），Spark SQL写入Hive的流程如下：

（1）Spark写入Hive会先生成一个临时的_temporary目录用于存储生成的数据文件，全部生成完毕后全部移动到输出目录，然后删除_temporary目录，最后创建Hive元数据
（2）多个Spark写入数据任务使用了同一个_temporary目录，导致其中一个完成数据生成和移动到Hive路径之后删除_temporary目录失败（因为还有其他Spark任务在往里面写），进一步导致数据已经到了但是元数据没有创建
（3）上一个任务虽然生成了数据文件但是没有元数据，则后一个任务的overwrite找不到元数据因此无法删除Hive路径下的数据文件
（4）当最后一个执行完成的Spark插入任务结束后，此时Hive路径下已经移动过来多个任务的数据文件，由于已经没有正在执行的Spark写任务，因此删除_temporary目录成功，创建元数据成功，结果就是这个元数据对应了该Hive路径下所有版本的数据文件。

最后的话：

我的场景是全量覆盖，因为人为重复设置定时任务原因导致数据翻倍，仅仅是查明了原因所在。若小伙伴有业务需求需要同时两个SparkSql任务写入一个Hive表，可采用创建Hive分区表，同时执行两个SparkSQL任务，往同一个Hive表分区插入，结果正常。

create table test_spark
(
`id` string,
`name` string
)
PARTITIONED BY (
`dt` string
)
STORED AS PARQUET;

查阅资料网上还有其他类似场景：

某张表的分区下的写入任务之前曾用SparkSQL执行过，跑失败后切换成HiveSQL执行成功了。查看该分区对应的目录，发现目录下同时存在SparkSQL和Hive SQL生成的文件（也就是Hive任务执行时未删除旧的sparkSQL的数据文件），因此导致查询有数据重复。

Hive 执行overwrite语句时是否删除旧数据机制：

当Hive执行insert overwrite写入数据到分区时，根据分区的元数据是否存在来决定是否要清空分区目录下的所有文件：

如果分区元数据存在（HiveMetaStore中有分区记录），则清空分区下的所有元数据
如果分区元数据不存在（仅针对外部表），Hive不会去自动推测分区对应的路径，也就不会去删除该分区下的所有文件.

排查HiveMetaStore的元数据，发现该分区的创建时间是在Hive SQL执行完后，可以推测出之前失败的SparkSQL任务虽然生成了数据文件，但是未生成对应的Hive元数据，因此出现这种情况。

SparkSQL 失败的原因

在Spark执行任务时，会创建一个临时目录，这个临时目录路径为 ${outputPath}/_temporary (outputPath为任务设置的OutputFormat的outputPath)。spark会将执行过程中生成的文件先落地到临时目录中，最终任务执行成功了才全部移动到最终的输出目录。最后，这个临时目录会在任务执行结束后被删除.

经测试发现，SparkSQL执行某张表的分区写入时，它生成的临时目录位于表路径下。如果这张表不是ORC或者Parquet表，它的临时目录就和Hive比较像，如/a/test/.hive-staging_hive_2020-10-23_16-41-55_549_7302943708666306032-5 (/a/test为表路径)，如果这张表是ORC或者Parquet表，sparkSQL生成的临时目录就变成/a/test/_temporary。

上面业务写入的那张表是ORC表，因此如果有多个任务在同时写入，就会有问题：任务结束时spark要删除掉这个临时目录，而其他的任务也在使用这个临时目录,所以SparkSQL任务的错误原因就可以理通了：

比如任务A执行结束，要删除/a/test/_temporary这个临时目录，但是任务B还在执行，下面还有任务B的一些临时文件存在，这时候任务A执行的删除操作就会报错（如同Excel等文件正在使用过程中你没法修改文件名）提示临时目录下还有其他的文件，因此无法删除临时目录。同时，SparkSQL在删除完临时目录后才会添加Hive元数据，因此这里删除临时目录失败就导致了后面的数据重复的问题。

Hive在写入数据的时候也会创建临时目录，但是在非动态分区的写入模式下，Hive创建的临时目录是在具体的分区路径下，比如/user/warehouse/dt=20201022，因此各个分区的写入任务是可以同时并行的。

SparkSQL并发写入orc、parquet表的异常问题排查_u013332124的专栏-CSDN博客

Hive “INSERT OVERWRITE” Does Not Remove Existing Data – Hadoop Troubleshooting Guide – Eric's Blog

Spark任务输出文件过程详解_u013332124的专栏-CSDN博客_spark输出到一个文件

SparkSQL overwrite插入Hive表数据重复问题相关推荐

spark用scala读取hive表数据（不同版本区别）
spark用scala读取hive表数据 spark1.6写法: val conf = new SparkConf() val sc = new SparkContext(conf) ...
hive 如何将数组转成字符串_hive 将hive表数据查询出来转为json对象和json数组输出...
1.将hive表数据查询出来转为json对象输出apache 一.将查询出来的数据转为一行一行,并指定分割符的数据json 二.使用UDF函数,将每一行数据做为string传入UDF函数中转换为jso ...
sqoop导出数据单mysql_sqoop导出hive表数据到mysql
直接在mysql里从本地文件系统导入数据 mysql>LOAD DATA LOCAL INFILE 'C:\\Users\\asys\\Documents\\Tencent Files\\131 ...
[源码分享] HIVE表数据量统计邮件
概要: 计算HIVE BI库下每天数据表总大小及增量输出: 总大小:xxxG 日同比新增数据量:xxxG 周同比新增数据量:xxxG 月同比新增数据量:xxxG总表数:xxx 日新增表数:xxx 周 ...
Hive表数据量统计原理和源码分析
在Hive explain获得执行计划时,经常会看到如下图所示的表数据量统计: 那么这个数据量,Hive是如何统计出来的呢? 一.Data size统计 1.1.Hive源码在Hive通过Antlr ...
navicat批量插入mysql表数据
打开Navicat的查询,写入以下代码,定义一个方法: delimiter // CREATE PROCEDURE pro_insertMany(in num INT) BEGINDECLARE i ...
oracle怎么批量执行tdl文件,ORACLE no1 存储过程插入更新表数据
CREATE OR REPLACE PROCEDURE sp_cust_main_data_yx(InStrDate IN VARCHAR2, OS_ERR_MSG OUT VARCHAR2) AS ...
基于K8S，spark访问hadoop集群的共享层hive表数据历险记
前言由于Spark的抽象设计,我们可以使用第三方资源管理平台调度和管理Spark作业,比如Yarn.Mesos和Kubernetes. 基本原理基本原理当我们通过spark-submit将Spar ...
MDG：通过API插入MDG表数据
代码参考: 存在指针重复利用的时候,需要清空指针,否则会出现各种问题 *&----------------------------------------------------------- ...

SparkSQL overwrite插入Hive表数据重复问题

SparkSQL overwrite插入Hive表数据重复问题相关推荐

最新文章

热门文章