跳过的阶段对 Spark 作业有任何性能影响吗？

Question

我正在运行一个 Spark 结构化流式传输作业，该作业涉及创建一个空数据帧，并使用每个微批处理对其进行更新，如下所示。每执行一次微批处理，阶段数就会增加 4。为了避免重新计算，我在循环内的每次更新后将更新的 StaticDF 持久化到内存中。这有助于跳过每个新微批次创建的额外阶段。

我的问题 -

1) 即使完成的阶段总数保持不变，因为增加的阶段总是被跳过，但它是否会导致性能问题，因为在一个时间点可能有数百万个跳过的阶段？
2）当缓存的RDD的一部分或全部不可用时会发生什么？ （节点/执行器故障）。 Spark 文档说到目前为止它并没有具体化从多个微批次接收到的全部数据，这是否意味着它需要再次从 Kafka 读取所有事件以重新生成 staticDF？

// one time creation of empty static(not streaming) dataframe
val staticDF_schema = new StructType()
      .add("product_id", LongType)
      .add("created_at", LongType)
var staticDF = sparkSession
.createDataFrame(sparkSession.sparkContext.emptyRDD[Row], staticDF_schema)

// Note : streamingDF was created from Kafka source
    streamingDF.writeStream
      .trigger(Trigger.ProcessingTime(10000L))
      .foreachBatch {
        (micro_batch_DF: DataFrame) => {

        // fetching max created_at for each product_id in current micro-batch
          val staging_df = micro_batch_DF.groupBy("product_id")
            .agg(max("created").alias("created"))

          // Updating staticDF using current micro batch
          staticDF = staticDF.unionByName(staging_df)
          staticDF = staticDF
            .withColumn("rnk",
              row_number().over(Window.partitionBy("product_id").orderBy(desc("created_at")))
            ).filter("rnk = 1")
            .drop("rnk")
              .cache()

          }

Answer 1

即使跳过的阶段不需要任何计算，但我的工作在一定数量的批次后开始失败。这是因为每次批处理执行时 DAG 都会增长，使其无法管理并引发堆栈溢出异常。

为了避免这种情况，我不得不打破火花谱系，这样每次运行的阶段数都不会增加（即使它们被跳过）