为什么spark流示例中的StatefulNetworkWordCount需要HashPartioner？

Question

我的问题是关于StatefulNetworkWordCount的例子:

https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/StatefulNetworkWordCount.scala

Q1）stateDstream RDD由驱动程序或工作节点维护，还是每个工作节点都有自己的完整状态rdd的本地副本？

Q2)为什么我们在下面的代码中需要一个HashPartitioner:

val stateDstream=wordDstream。updateStateByKey[Int]（newUpdateFunc，新HashPartitioner（ssc.sparkContext.defaultParallelism），true，initialRDD）

这里的幕后发生了什么？

Answer 1

要回答您的两个问题：

1)由< code>DStream生产的< code>RDD被分配给工人。与非流式相似，这意味着由< code>DStream生成的每个< code>RDD的记录分布在整个集群中(这就是分区在这里很重要的原因)。

2）在这种情况下，分区很重要，因为它确定了每次RDD迭代的记录是如何拆分的。尤其是像updateStateByKey（）这样的转换，您倾向于看到RDD在不同批处理间隔中的键保持不变。所以这里不用说，如果我们来自每个间隔RDD的键排列在相同的分区中，这个函数可以更有效地工作，并且可以更新分区中键的状态。

例如，让我们看一下您链接的字数统计程序。让我们考虑两个一秒间隔的RDD（rdd1在t= 1和rdd2在t = 2）。假设生成的rdd1用于文本“你好世界”，而生成的rdd2也看到文本“你好，我是世界”。如果不进行分区，每个RDD的记录可以发送到各个工作线程上的各种分区（t=1处的“hello”和t=2处的“hello”可以发送到不同的位置）。这意味着对计数状态的更新需要在每次迭代时重新排列记录以获取更新的计数。定义分区程序后（并记住其中一个参数所指示的！），我们将在同一分区上看到键“hello”和“world”，从而避免洗牌，并创建更有效的更新。

这里还必须注意，由于键可以更改，因此有一个参数可以切换是否记住分区程序。