触发一次的火花流追加输出模式

Question

null

触发器是否支持一次追加模式？

这里有一个最小的应用程序来再现这个问题。要旨

case class Model(valueForGroupBy: Int, time: Timestamp)

object Application {

  val appName = "sample"

//  val outputMode: OutputMode = OutputMode.Complete() // OK
  val outputMode: OutputMode = OutputMode.Append() // KO with trigger once

  val triggerMode: Trigger = Trigger.Once()

  val delayThreshold: FiniteDuration = 1.minute // watermarking wait for late

  val duration : FiniteDuration = 1.minute // window duration and slide

  val topic = "SAMPLE_TOPIC"
  val bootstrapServers = "127.0.0.1:9092"


  type KafkaKV = (String, String)

  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder
      .appName(appName)
      .getOrCreate()

    import org.apache.spark.sql._
    import org.apache.spark.sql.functions._
    import spark.implicits._

    val streamReader: DataStreamReader = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", bootstrapServers)
      .option("subscribe", topic)
      .option("startingOffsets", "earliest")
      .option("failOnDataLoss", "false")

    val df: DataFrame = streamReader.load()

    val ds: Dataset[Model] = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)").as[KafkaKV]
      .select(from_json($"value", Encoders.product[Model].schema).as("json"))
      .select($"json.*")
      .as[Model]

    val groupByColumns = Seq(
      window(new Column("time"), windowDuration = duration.toString, slideDuration = duration.toString),
      new Column("valueForGroupBy")
    )

    val agg = ds
      .withWatermark("time", delayThreshold.toString)
      .groupBy(groupByColumns: _*)
      .count()

    val streamWriter = agg
      .selectExpr(s"CAST(valueForGroupBy AS STRING) AS key", "to_json(struct(*)) AS value")
      .writeStream
      .trigger(triggerMode)
      .outputMode(outputMode)
      .format("console")
      .option("truncate", value = false)

    val streamingQuery = streamWriter.start()

    streamingQuery.awaitTermination()

  }
}

Answer 1

这是Spark已知的BUG。

在实际实现中，水印是在下面的批处理中持久化的。由于触发一次时没有后续批处理，因此watermak永远不会持久化。

Spark bugtracker中存在票据：https://issues.apache.org/jira/browse/Spark-24699