美文网首页
Spark踩坑vlog——使用外部存储减轻join时数据负担

Spark踩坑vlog——使用外部存储减轻join时数据负担

作者: 小段DSH12138 | 来源:发表于2021-07-25 21:39 被阅读0次

背景

        此次需要将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调度时间;但是相比于同资源下数据跑不出来,这些延迟都是可以接受的。
        此坑完结,撒花🎉~

相关文章

网友评论

      本文标题:Spark踩坑vlog——使用外部存储减轻join时数据负担

      本文链接:https://www.haomeiwen.com/subject/okyimltx.html