Spark 跨接收器的结构化流一致性答案

【问题标题】：Spark structured streaming consistency across sinksSpark 跨接收器的结构化流一致性
【发布时间】：2018-04-19 23:00:01
【问题描述】：

我想在以下情况下更好地理解 Spark 2.2 结构化流的一致性模型：

一个来源（Kinesis）
从该源向 2 个不同的接收器进行 2 次查询：一个用于存档目的的文件接收器 (S3)，另一个用于处理数据的接收器（数据库或文件，尚未确定）

我想了解是否有跨接收器的一致性保证，至少在某些情况下：

一个水槽可以领先于另一个吗？还是他们在源上以相同的速度消耗数据（因为它的源相同）？它们可以同步吗？
如果我（优雅地）停止流应用程序，两个接收器上的数据是否一致？

原因是我想构建一个类似Kappa 的处理应用程序，能够在我想重新处理某些历史记录时暂停/关闭流式传输部分，并且当我恢复流式传输时，避免重新处理某些内容已经被处理过（在历史中），或者丢失了一些（例如，一些尚未提交到存档的数据，然后在流媒体恢复时跳过已处理）

【问题讨论】：

我在同一条船上。您是否找到了保持两个水槽之间一致性的解决方案？如果是的话，你能分享你的方法吗..！谢谢

标签： apache-spark spark-streaming

【解决方案1】：

要记住的重要一点是 2 个接收器将用于 2 个不同的查询，每个查询都独立于源读取。所以检查点是按查询完成的。

每当您在DataStreamWriter 上调用start 时都会产生新查询，如果您设置checkpointLocation，每个查询都会有自己的检查点来跟踪与接收器的偏移量。 p>

val input = spark.readStream....

val query1 = input.select('colA, 'colB)
  .writeStream
  .format("parquet")
  .option("checkpointLocation", "path/to/checkpoint/dir1")
  .start("/path1")

val query2 = input.select('colA, 'colB)
  .writeStream
  .format("csv")
  .option("checkpointLocation", "path/to/checkpoint/dir2")
  .start("/path2")

因此，每个查询都从源读取并独立跟踪偏移量。这也意味着，每个查询可以位于输入流的不同偏移量处，您可以重新启动其中一个或两个而不影响另一个。

更新

我想提出另一个建议，因为Databricks Delta 是开源的。我使用的一种常见模式是将来自上游源的数据直接登陆到仅附加的 Delta 表中。然后，使用结构化流，您可以高效地订阅表并以增量方式处理新记录。 Delta 的内部事务日志比基本文件源所需的 S3 文件列表更有效。这可确保您在多个查询中拥有一致的数据源，从 S3 与 Kinesis 拉取。

【讨论】：

谢谢@Silvio。顺便说一句，通过您的示例，我发现有关检查点的一个棘手问题：如果通过option() 指定，则需要手动拆分dir1 和dir2，而如果使用spark.sql.streaming.checkpointLocation 全局指定，则附加一个后缀（@ 987654331@ 或 UUID）。参考Spark code
@mathieu 很高兴为您提供帮助！你能接受它作为答案吗？

【解决方案2】：

Silvio 所写的内容绝对正确。写入 2 个接收器将启动两个相互独立运行的流式查询（实际上是 2 个流式应用程序读取相同的数据 2 次并处理 2 次并自行检查点）。

我想补充一点，如果您希望两个查询同时停止/暂停以防任何一个查询重新启动或失败，则可以选择使用 api：awaitAnyTermination em>()

而不是使用：

query.start().awaitTermination()

使用：

sparkSession.streams.awaitAnyTermination()

添加 api 文档的摘录：

/**
   * Wait until any of the queries on the associated SQLContext has terminated since the
   * creation of the context, or since `resetTerminated()` was called. If any query was terminated
   * with an exception, then the exception will be thrown.
   *
   * If a query has terminated, then subsequent calls to `awaitAnyTermination()` will either
   * return immediately (if the query was terminated by `query.stop()`),
   * or throw the exception immediately (if the query was terminated with exception). Use
   * `resetTerminated()` to clear past terminations and wait for new terminations.
   *
   * In the case where multiple queries have terminated since `resetTermination()` was called,
   * if any query has terminated with exception, then `awaitAnyTermination()` will
   * throw any of the exception. For correctly documenting exceptions across multiple queries,
   * users need to stop all of them after any of them terminates with exception, and then check the
   * `query.exception()` for each query.
   *
   * @throws StreamingQueryException if any query has terminated with an exception
   *
   * @since 2.0.0
   */
  @throws[StreamingQueryException]
  def awaitAnyTermination(): Unit = {
    awaitTerminationLock.synchronized {
      while (lastTerminatedQuery == null) {
        awaitTerminationLock.wait(10)
      }
      if (lastTerminatedQuery != null && lastTerminatedQuery.exception.nonEmpty) {
        throw lastTerminatedQuery.exception.get
      }
    }
  }

【讨论】：