在合理的时间内找到两个大数据集中的共同元素

Posted 2023-04-17

【中文标题】在合理的时间内找到两个大数据集中的共同元素

【英文标题】：Find Common Elements in two Big Data Set in a Reasonable Time

【发布时间】：2018-04-19 08:58:49

【问题描述】：

我有两个 Spark 数据框 DFa 和 DFb，它们具有相同的架构（'country'、'id'、'price'、'name'）。

DFa 有大约 6.1 亿行， DFb 有 30 亿行。

现在我想从 DFa 和 DFb 中找到所有具有相同 id 的行，其中 id 看起来像“A6195A55-ACB4-48DD-9E57-5EAF6A056C80”。

这是一个 SQL 内连接，但是当我运行 Spark SQL 内连接时，一个任务被杀死，因为容器使用了太多内存并导致 Java 堆内存错误。而我的集群资源有限，调优YARN和Spark配置不是一个可行的方法。

还有其他解决方案可以解决这个问题吗？如果运行时可以接受，不使用 spark 解决方案也是可以接受的。

更一般地说，在两个非常大的数据集中找到共同元素时，谁能给出一些算法和解决方案。

【问题讨论】：

连接两个数据集（可能只有ID）并通过分组提取具有相同ID的数据集是否可以接受？

一个数据库，两张表，使用id作为主键，获取DFb的最小和最大id范围，一个动态查询，获取DFb的id范围集并与DFa进行比较，然后插入第三张表.编写一个 Java 方法以合理的长度遍历 id 范围，每次迭代后关闭/重新打开数据库连接以释放资源。

我不知道 spark，但是一个很好的通用方法是根据 ID 的哈希将记录分区到存储桶中，然后在每个存储桶中独立地进行联接。

当数据库崩溃时，我很幸运能将这种大小的数据集提取到文件中，然后使用 unix 排序实用程序将匹配的 ID 放在一起，然后按顺序处理。就像@ChatterOne 推荐的那样。

【参考方案1】：

首先计算您的 ID 的 64 位哈希值。在哈希上的比较会比在字符串 id 上快得多。

我的基本想法是：

从 DFa 构建哈希表。 在计算 DFb 的哈希值时，您会在表中进行查找。如果那里什么都没有，则删除条目（不匹配）。如果您成功了，请比较实际的 ID，以确保您没有误报。

复杂度为 O(N)。不知道您期望有多少重叠，这是您能做的最好的事情，因为您可能必须输出所有内容，因为它们都匹配。

简单的实现将使用大约 6GB 的内存用于表（假设 80% 的占用率并且您使用平面哈希表），但您可以做得更好。 由于我们已经有了哈希，我们只需要知道它是否存在。因此，您只需要一位来标记可以大大减少内存使用的内容（每个条目需要减少 64 倍的内存，但您需要降低占用率）。但是，这不是一个常见的数据结构，因此您需要实现它。

但还有更好的东西，更紧凑的东西。这就是所谓的布隆过滤器。这将引入更多误报，但无论如何我们都必须仔细检查，因为我们不信任哈希，所以这不是一个很大的缺点。最好的部分是您应该能够找到已经可用的库。

所以所有东西看起来像这样：

从 DFa 计算哈希并构建布隆过滤器。 从 DFb 计算哈希值并检查布隆过滤器。如果您在 DFa 中获得匹配查找 ID，以确保它是真正的匹配并将其添加到结果中。

【讨论】：

谢谢！我用谷歌搜索并找到了一个博客。 bloom-filter-assisted-joins-using-pyspark 可能会有所帮助

【参考方案2】：

这是任何大数据环境中的典型用例。您可以使用 Map-Side 连接来缓存广播给所有执行程序的较小表。

您可以在此处阅读有关广播加入的更多信息

Broadcast-Joins

【讨论】：

谢谢。对于小数据集（默认 spark.sql.autoBroadcastJoinThreshold=10M），广播是可行的。但是两者较小的数据集大约是10G+

在这种情况下，您可以对表进行分区，然后进行广播连接。加入所有数据后，您可以使用 Union。