问题标签 [partitioner]

我有一个数据集，我想将其写入 parquet 文件，以便之后通过 Spark 请求这些文件，包括 Predicate Pushdown。

目前我使用按列重新分区和分区数将数据移动到特定分区。该列正在标识相应的分区（从 0 开始到（固定）n）。结果是 scala/spark 产生了意想不到的结果并创建了更少的分区（其中一些是空的）。也许是哈希冲突？

为了解决这个问题，我试图找出原因并试图找到解决方法。我通过将数据帧转换为 rdd 并将 partitionBy 与 HashPartitioner 一起使用找到了一种解决方法。令我惊讶的是：我得到了预期的结果。但是将数据帧转换为 RDD 对我来说不是一个解决方案，因为它占用了太多资源。

我已经测试过这个环境

• Cloudera CDH 5.9.3 上的 SPARK 2.0

• emr-5.17.0 上的 SPARK 2.3.1

这是我的输出测试。请使用 Spark-shell 运行它们

我首先通过 HashPartitioner 被用于对数据帧进行重新分区的方法，但事实并非如此，因为它正在处理 RDD。

谁能指导我这个“Exchange hashpartitioning”（见上面的解释输出）是如何工作的？

2019-01-16 12:20：这不是HashPartitioner如何工作的副本？因为我对 Integer 列上按列（+ 数字分区）重新分区的哈希算法感兴趣。如您在源代码中所见，一般 HashPartitioner 正在按预期工作。

为了减少两个 RDD 加入过程中的洗牌，我决定先使用 HashPartitioner 对它们进行分区。这是我的做法。我做得对吗，还是有更好的方法来做到这一点？

我有一个包含 10 个分区的主题，并且我生成了具有A,B,C,D,E,F,G,H,I9 个不同键的事件。

我观察到这样做的消息：

同一分区中有 2 条具有不同键的消息，并且也有空分区。

Kafka的默认分区器是否会产生冲突？

我从一个平衡到两个默认休息生产者的流进行生产。

这是我所期待的：

我最近阅读了一篇文章，描述了如何自定义分区数据帧 [ https://dataninjago.com/2019/06/01/create-custom-partitioner-for-spark-dataframe/ ]，其中作者用 Python 说明了该技术. 我使用 Scala，该技术看起来是解决倾斜问题的好方法，所以我尝试了类似的方法，我发现当一个人执行以下操作时：

我们注意到以下结果：

所以，我得出了以下结论……这实际上对我有用……但我无法理解使用 Python 的文章中的行为，这让我很恼火：

现在......我希望我错了！也许在 Spark 内部有更多专业知识的人可以在这里指导我。我写了一个小程序（如下）来说明结果。如果你能让我直截了当，请提前致谢。

更新

除了说明问题的 Spark 代码之外，我还尝试了 Python 中原始文章的简化版本。下面的转换创建一个数据帧，提取其底层 RDD 并对其重新分区，然后恢复数据帧并验证分区器是否丢失。

说明问题的 Python 片段

说明问题的 Scala 片段

有没有人在强度比较器类的同时尝试过 Optaplanner 的分区搜索功能？

首先，我创建了一个自定义分区器，它拆分计划实体并分配计划值（它不拆分计划值，因为我需要每个分区的所有值）。到目前为止效果很好。

然后我尝试实现规划值强度，如果我不使用分区器但代码在我使用时停止运行，它似乎可以解决。任何想法为什么会这样？

我正在尝试使用 Kafka 的 RoundRobinPartitioner 类在所有分区之间均匀分布消息。我的Kafka主题配置如下：

名称：multischemakafkatopicodd

分区数：16

复制因子：2

比如说，如果我产生 100 条消息，那么每个分区应该有 6 或 7 条消息。但是，我得到了类似的东西：

我认为可能是我没有产生足够的消息，所以我尝试了 1M 记录并将分区数设置为奇数：

主题：multischemakafkatopicodd

分区数：31

复制因子：2

...我得到了这个。这一次，每个分区中的消息数量有点均匀分布。

我再次进行了相同的测试，但将分区数量减少到 8 个，我得到了这个结果，我们可以清楚地看到一些分区有接近 15K 的消息，而其他分区有大约 10K：

我做错了什么还是它应该如何工作？为什么消息的分布如此不均？

如果有人可以帮助我，那就太好了。谢谢。