星火流-> DStream.checkpoint与SparkStreaming.checkpoint

Question

我有Spark1.4流应用程序，它从Kafka读取数据，使用状态转换，并有15秒的批处理间隔。

为了使用状态转换，以及从驱动程序故障中恢复，我需要在流上下文上设置检查点。

此外，在Spark1.4文档中，他们建议DStream检查点是批处理间隔的5-10倍。

所以我的问题是：

如果我只在星火流上下文上设置检查点会发生什么？我想DStreams每批间隔都会被检查？

如果我同时设置流上下文上的检查点以及从Kafka读取数据的那一刻，我设置：

DStream.checkpoint(90秒)

元数据检查点的间隔是什么，数据检查点的间隔是什么(意思是DStreams)？

谢谢。

Answer 1

我想DStreams每批间隔都会被检查？

不，星火将检查您的数据每批间隔乘以一个常数。这意味着，如果批处理间隔为15秒，数据将每15秒多次检查一次。例如，在mapWithState中，它是一个有状态流，您可以看到批处理间隔乘以10：

private[streaming] object InternalMapWithStateDStream {
  private val DEFAULT_CHECKPOINT_DURATION_MULTIPLIER = 10
}

元数据检查点的间隔是什么，数据检查点的间隔是什么(意思是DStreams)？

如果您将DStream上的检查点持续时间设置为90秒，那么这将是您的检查点持续时间，这意味着每90秒数据将得到检查点。您不能直接在StreamingContext上设置检查点持续时间，您能做的就是传递检查点目录。checkpoint的重载只需要一个String

/**
 * Set the context to periodically checkpoint the DStream operations for driver
 * fault-tolerance.
 * @param directory HDFS-compatible directory where the checkpoint
 *        data will be reliably stored.
 *        Note that this must be a fault-tolerant file system like HDFS.
 */
def checkpoint(directory: String)

编辑

对于updateStateByKey，检查点的时间似乎设置为批处理时间乘以Seconds(10) / slideDuration。

// Set the checkpoint interval to be slideDuration or 10 seconds,
// which ever is larger
if (mustCheckpoint && checkpointDuration == null) {
  checkpointDuration = slideDuration * math.ceil(Seconds(10) / slideDuration).toInt
  logInfo(s"Checkpoint interval automatically set to $checkpointDuration")
}