Spark mapWithState 将所有数据洗牌到一个节点

Posted 2023-04-15

【中文标题】Spark mapWithState 将所有数据洗牌到一个节点

【英文标题】：Spark mapWithState shuffles all data to one node

【发布时间】：2016-03-22 10:04:39

【问题描述】：

我正在开发一个 Scala (2.11) / Spark (1.6.1) 流式传输项目，并使用 mapWithState() 跟踪以前批次的数据。

状态分为 20 个分区，由 StateSpec.function(trackStateFunc _).numPartitions(20) 创建。我曾希望将状态分布在整个集群中，但似乎每个节点都拥有完整的状态，并且执行总是只在一个节点上执行。

Locality Level Summary: Node local: 50 显示在每个批次的 UI 中，完整的批次是随机读取的。之后，我写入 Kafka，分区再次分布在集群中。我似乎无法找出为什么 mapWithState() 需要在单个节点上运行。如果状态受限于一个节点而不是整个集群，这是否会破坏状态分区的概念？难道不能通过key来分发状态吗？

【问题讨论】：

我认为您应该添加您的 spark DAG，并添加更多关于您使用 mapWithState 的类型的详细信息。

【参考方案1】：

我似乎无法找出为什么 mapWithState 需要在 单节点

它没有。 Spark 默认使用HashPartitioner 在集群中的不同工作节点之间划分密钥。如果由于某种原因您看到所有数据都存储在不同的节点上，请检查密钥的分布。如果这是您用作键的自定义对象，请确保它的 hashCode 方法已正确实现。如果密钥分配有问题，就会发生这种情况。如果您想对此进行测试，请尝试使用随机数作为您的密钥并查看 Spark UI 并查看此行为是否发生变化。

我正在运行 mapWithState 并且传入的数据基于密钥进行分区，因为在保持状态之前以及查看 Spark 上的 Storage 选项卡时，我还有一个 reduceByKey 方法调用UI，我可以看到不同的 RDD 存储在集群的不同工作节点上。

【讨论】：

我的数据通过 kafka 进入由密钥分区的 spark 流，但我有同样的问题。你介意扩大你的答案吗？如何访问/编辑哈希分区器，以及如何在写入数据库之前应用 reduceByKey？

@Andreas 您遇到了什么问题？分区偏差？

感谢您的回复，我想学习如何控制 kafka 和 spark 的 key 分区，因为在我的问题中，数据不需要洗牌。在实践中，我观察到改组，因为应用于我的复合键的哈希函数会将数据发送出去。下面是我的问题的详细信息。

我遇到了展示次数问题。 Kafka 按输入类（一个数字，例如 URL ID#）进行分区并传输一对（URLID、时间戳）。我将时间戳四舍五入到最接近的时间间隔并形成一个键“URLID;RoundedTimeStamp”，然后进行累积字数。 DAG 会为每个 RDD 打乱，所以我认为我必须提供一个 HashPartitioner，它单独为 URLID 和复合键“URLID;RoundedTimeStamp”返回相同的值，理想情况下也已经在 kafka 端以相同的方式进行分区。我一直在寻找 updateStateByKey 示例，但希望使用 mapWithState 并且不确定语法

@Andreas Spark 默认使用HashPartitioner 进行需要随​​机播放的操作，因此无需明确提供。关于mapWithState语法，see this blog post（免责声明：我是作者）。

【参考方案2】：

你在 --deploy-mode 集群上运行 spark 吗？请检查一下。

还要确保您设置了 --num-executors 20 --executor-cores 10 因为除非您在默认情况下使用动态分配运行，否则它将分配 2 个执行器。