背景
此次需要将10张表[A、B、C、D、E、F、G、H、I、J]的数据union与表∂中的一个字段进行join,以达到筛选数据的目的;其中表A的数据量约为320亿,其余9个表各自数据量约为20亿,表∂的数据量约为900万;
资源使用情况上升至100 executor-nums * 3 executor-core * 20G executor-memory,但还是卡在join的stage上,重复失败多次后,整个job失败;
实现方法
方法1
val tableArr = Array("table_a_name", "table_b_name", "table_c_name", "", "table_d_name", "table_e_name", "table_f_name", "table_g_name", "table_h_name", "table_i_name", "table_j_name")
val data_df = tableArr.map(table_name =>
spark.sql(
s"""
|SELECT a, b, c
|FROM ${table_name}
|""".stripMarigin)
).reduce(_.union(_))
val filter_table_tmp = spark.sql(
s"""
|SELECT a
|FROM ∂
|""".stripMarigin)
val result = data_df.join(filter_table_tmp, Seq("a"), "inner")
采用上面这种代码,通过查看作业的DAG图,发现程序将前十个表的数据union到一起后,再与表∂进行join;相当于将400多亿与900万直接进行join;
老办法,1.将spark.sql.shuffle.partition的值调整到程序总executor-core的2~3倍;2.检查join的字段在两个表中的类型是否相同;但是最终还是执行失败;
一直报org.apache.spark.shuffle.FetchFailedException
这类错误;果然,还是shuffle的时候,数据量太大/task多/一个task执行的数据太多的问题;
如果,不想采用本次标题的方法解决问题的同学,可以看看我之前的博客:Spark踩坑vlog——join时shuffle的大坑,可能会帮助你解决问题;
方法2[失败]
最终目的还是想将数据打散、分开计算,所以,就想把这几个表一个一个进行join,再将join的结果union到一起,代码:
val filter_table_tmp = spark.sql(
s"""
|SELECT a
|FROM ∂
|""".stripMarigin).cache()
val table_a_df = spark.sql(
s"""
|SELECT a, b, c
|FROM table_a_name
|""".stripMarigin)
val table_a_result_df = table_a_df.join(filter_table_tmp, Seq("a"), "inner")
val table_b_df = spark.sql(
s"""
|SELECT a, b, c
|FROM table_b_name
|""".stripMarigin)
val table_b_result_df = table_b_df.join(filter_table_tmp, Seq("a"), "inner")
......
val result = table_a_result_df.union(table_b_result_df)......
本以为采用以上代码,可以把每个表的join分别计算,结果谁承想,通过查看DAG图发现,Spark在执行的时候,发现了我们的join都拿filter_table_tmp进行的,结果跟方法1一样,先把所有数据合并到一起,然后再进行join;
因为无知,方法2失败。。。。。。
方法3
那么,最终如果想达到每个表分开join的目的,只能每次join用触发为一个job,为了方便之后的数据校验,我选择直接将中间数据写入外部存储。不太懂Spark的job、task啥划分的同学,可以看看我之前的博客Spark的组件们[Application、Job、Stage、TaskSet、Task]。
val filter_table_tmp = spark.sql(
s"""
|SELECT a
|FROM ∂
|""".stripMarigin).cache()
val table_a_df = spark.sql(
s"""
|SELECT a, b, c
|FROM table_a_name
|""".stripMarigin)
table_a_df.join(filter_table_tmp, Seq("a"), "inner").createOrReplaceTempView("table_a_result_tmp")
spark.sql(
s"""
|INSERT OVERWRITE TABLE temporary_storage_table_name PARTITION(event = "a")
|SELECT *
|FROM table_a_result_tmp
|""".stripMarigin)
......
val result = spark.sql(
s"""
|SELECT *
|FROM temporary_storage_table_name
|""".stripMarigin)
最后,一个表join生成一个job,除了表A执行时数据大,耗时比较久一点外,数据可算是跑出来了。
因为程序按照job顺序执行,所以在每次对一个表进行join时,程序所有的资源都会用来执行一个表的数据,降低了数据量,完成了"少量多次"的理念。缺点就是,job太多,采用了外部存储,增加了IO写入读取时间 & 增加了job调度时间;但是相比于同资源下数据跑不出来,这些延迟都是可以接受的。
此坑完结,撒花🎉~
网友评论