Spark Streaming 动态分配执行器AllocationManager答案

【问题标题】：Spark Streaming Dynamic Allocation ExecutorAllocationManagerSpark Streaming 动态分配执行器AllocationManager
【发布时间】：2017-03-22 21:49:03
【问题描述】：

我们有一个带有mapWithState 的spark 2.1 流应用程序，启用spark.streaming.dynamicAllocation.enabled=true。流水线如下：

var rdd_out = ssc.textFileStream()
    .map(convertToEvent(_))
    .combineByKey(...., new HashPartitioner(partitions))
    .mapWithState(stateSpec)
    .map(s => sessionAnalysis(s))
    .foreachRDD( rdd => rdd.toDF().....save(output));

流式应用程序从 2 个执行程序开始，一段时间后它会创建新的执行程序，因为负载会按预期增加。问题是这些执行者不共享负载。

Partitions 的数量大到足以溢出到新的 executors，并且 key 是均匀分布的，我设置了 40+ 个 partitions，但我只能看到 8 个 partitions（2 executors x 4 cores each） mapWithState 存储。我期待在分配新的执行程序时，这 8 个分区会被拆分并分配给新的分区，但这永远不会发生。

请指教。

谢谢，

【问题讨论】：

我很确定有状态转换不能以这种方式工作，spark.streaming.dynamicAllocation.enabled 根本不会帮助你。 “状态”是这里的引用，因为它是分区的，所以它将作为一个“模板”。
@zero323，您能否详细说明“模板”，因为状态与任何其他状态一样保存在一组 rdd 上。我本来希望它可以重新平衡。谢谢
但要“重新平衡”，您必须重新洗牌，因此需要重新分区状态选项。

标签： apache-spark spark-streaming

【解决方案1】：

显然，答案一直盯着我的脸 :)。根据下面的文档，RDD 应该继承上游分区。

   * Otherwise, we use a default HashPartitioner. For the number of partitions, if
   * spark.default.parallelism is set, then we'll use the value from SparkContext
   * defaultParallelism, otherwise we'll use the max number of upstream partitions.

然而，mapWithState 中的状态没有上游 RDD。因此设置为默认并行度，除非您直接在状态中指定分区，如下例所示。

val stateSpec = StateSpec.function(crediting.formSession _)
        .timeout(timeout)
        .numPartitions(partitions)  // <----------

var rdd_out = ssc.textFileStream()
    .map(convertToEvent(_))
    .combineByKey(...., new HashPartitioner(partitions))
    .mapWithState(stateSpec)
    .map(s => sessionAnalysis(s))
    .foreachRDD( rdd => rdd.toDF().....save(output));

仍然需要弄清楚如何使分区数动态化，就像动态分配一样，这应该在运行时改变。

【讨论】：

“仍然需要弄清楚如何使分区数动态化，就像动态分配一样，这应该在运行时改变。”这也适用于流式应用程序中的任何其他 rdd。