apache-spark - 使用检查点 Spark Stream 的中流更改配置

Question

我有一个类似 this 的 Spark 流/DStream 应用程序:

// Function to create and setup a new StreamingContext
def functionToCreateContext(): StreamingContext = {
  val ssc = new StreamingContext(...)   // new context
  val lines = ssc.socketTextStream(...) // create DStreams
  ...
  ssc.checkpoint(checkpointDirectory)   // set checkpoint directory
  ssc
}

// Get StreamingContext from checkpoint data or create a new one
val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)

// Do additional setup on context that needs to be done,
// irrespective of whether it is being started or restarted
context. ...

// Start the context
context.start()
context.awaitTermination()

我的上下文使用配置文件，我可以在其中使用 appConf.getString 等方法提取项目。所以我实际上使用:

val context = StreamingContext.getOrCreate(
    appConf.getString("spark.checkpointDirectory"), 
    () => createStreamContext(sparkConf, appConf))

其中 val sparkConf = new SparkConf()...。

如果我停止我的应用程序并更改应用程序文件中的配置，除非我删除检查点目录内容，否则这些更改不会生效。例如，我想动态更改 spark.streaming.kafka.maxRatePerPartition 或 spark.windowDurationSecs。 (编辑:我终止应用程序，更改配置文件，然后重新启动应用程序。)我怎样才能动态更改这些设置或强制执行(已编辑的字词)配置更改而不破坏我的检查点目录(哪个即将包括状态信息的检查点)？

Answer 1

How can I do dynamically change these settings or enforce a configuration change without trashing my checkpoint directory?

如果深入了解 StreamingContext.getOrCreate 的代码:

def getOrCreate(
    checkpointPath: String,
    creatingFunc: () => StreamingContext,
    hadoopConf: Configuration = SparkHadoopUtil.get.conf,
    createOnError: Boolean = false
  ): StreamingContext = {
    val checkpointOption = CheckpointReader.read(
      checkpointPath, new SparkConf(), hadoopConf, createOnError)
    checkpointOption.map(new StreamingContext(null, _, null)).getOrElse(creatingFunc())
}

你可以看到，如果 CheckpointReader 在类路径中有检查点数据，它使用 new SparkConf() 作为参数，因为重载不允许传递自定义创建的 SparkConf。默认情况下，SparkConf 将加载声明为环境变量或传递给类路径的任何设置:

class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging {

  import SparkConf._

  /** Create a SparkConf that loads defaults from system properties and the classpath */
  def this() = this(true)

因此，实现您想要的目标的一种方法是，您可以通过 spark.driver.extraClassPath 和 传递参数，而不是在代码中创建 SparkConf 对象>spark.executor.extraClassPath 到 spark-submit。