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

背景

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

最后编辑于：2021.07.25 21:40:14