如何在配置单元表中检查损坏的记录

Question

我有一个配置单元表，数据每天递增。在某一天，一些损坏的记录被插入到表中。有没有一种方法可以将HDFS上的表与主文件进行匹配，并从Hive 中提取损坏的记录

或

如何在具有100万行的配置单元表中识别损坏的记录？

Answer 1

使用join, except1查找加载到配置单元表vs文件中的损坏记录。

Example:

//read the file
val df=spark.read.<format>("<path>")
//read hive table
val df1=spark.read.table("<db>.<hive_table_name>")
//without using md5 hash
df.exceptAll(df1).show()
df1.exceptAll(df).show()
//create md5 hash by concatenating all column values
val df2=df.withColumn("md_hash",md5(concat_ws(",",df.columns.map(c => col(c)): _*))).select("md_hash")
val df3=df1.withColumn("md_hash",md5(concat_ws(",",df.columns.map(c => col(c)): _*))).select("md_hash")
//get non matching rows from df2 that are not existed in df3
df2.except(df3).show()
df2.exceptAll(df3).show()
//get non matching rows from df3 that are not existed in df2
df3.exceptA(df2).show()
df3.exceptAll(df2).show()
//or using full outer join
df3.join(df2,df3("md_hash") === df2("md_hash"),"full").
filter((df2("md_hash").isNull || df3("md_hash").isNull)).
show(10,false)