Spark 平障录

Profile

Profile 是最重要的第一环。

• 利用好 spark UI 和 yarn container log
• 分析业务代码，对其计算代价进行预判
• 建设基准，进行对比，比如application id 进行对比，精确到 job DAG 环节

design 研究顺序

• paper => blog => code comment => code

充分利用 UI

Stage 页面

页头 summary，也很重要。

勾选所有 UI metrics，提供判断依据

task 维度 metrics

Executor 维度

application id 对比

如果两个application id， shuflfle read/write 差不多，input/output 差不多。
慢的那一个，如果内存消耗大，那说明计算量大，比如 sum， count 等

Executor 页面

选中所有 metrics，增加判断依据

注意GC

参考链接

https://www.jianshu.com/p/bd45b085da9a
https://medium.com/@suffyan.asad1/beginners-guide-to-spark-ui-how-to-monitor-and-analyze-spark-jobs-b2ada58a85f7

FQA

Job aborted due to stage failure: Total size of serialized results of 1936 tasks (20.0 GiB) is bigger than spark.driver.maxResultSize (20.0 GiB)

直接在sql 里 加上limit 10；

driver 主动cancel job 的场景

spark plan 在打开AQE 的过程中一直在变化。

运行完之后，又变成了 Final Plan

关闭AQE 就没有这个问题

set spark.sql.adaptive.enabled=false

最小模拟的case

set spark.sql.autoBroadcastJoinThreshold=-1;
select * from table1 where a='0' order by a;

即可复现 LocalTableScanExec，关键是模拟出shuffleExchange

或者以下UT也可以复现

  test("SPARK-35585: Support propagate empty relation through project/filter") {
    withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
      val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult(
        "SELECT key FROM testData WHERE key = 0 ORDER BY key, value")
      assert(findTopLevelSort(plan1).size == 1)
      assert(stripAQEPlan(adaptivePlan1).isInstanceOf[LocalTableScanExec])

      val (plan2, adaptivePlan2) = runAdaptiveAndVerifyResult(
       "SELECT key FROM (SELECT * FROM testData WHERE value = 'no_match' ORDER BY key)" +
         " WHERE key > rand()")
      assert(findTopLevelSort(plan2).size == 1)
      assert(stripAQEPlan(adaptivePlan2).isInstanceOf[LocalTableScanExec])
    }
  }