更新表时如何改进 Spark 中的 SQL 查询？ （子查询中的'NOT IN'）

Question

我在 Spark 中有一个 Dataframe，它注册为一个名为 A 的表，并且有 1 billion 记录和 10 列。第一列 (ID) 是主键。

还有另一个数据框，它注册为一个名为 B 的表，并且有 10,000 记录和 10 列（与表 A 相同的列，第一列 (ID) 是主键）。

表 B 中的记录是“更新记录”。所以我需要用表 B 中的记录更新表 A 中的所有 10,000 条记录。

我首先尝试了这个 SQL 查询：

select * from A where ID not in (select ID from B) 然后将其与表 B 联合。

方法可以，但第一次查询 (select * from A where ID not in (select ID from B)) 非常慢（中等集群上的小时数）。

然后我尝试使用 LEFT JOIN 加速第一个查询：
select A.* from A left join B on (A.ID = B.ID ) where B.ID is null

这种方法在逻辑上看起来不错，但对于 Spark 容器来说它需要大量内存
(YARN for exceeding memory limits. 5.6 GB of 5.5 GB physical memory used. Consider boosting spark.yarn.executor.memory)..

什么是更好/更快/更少内存消耗的方法？

Answer 1

我也会选择left join 而不是not in。

一些减少内存需求和性能的建议 -

1. 请看大表是按连接键（ID）均匀分布的。如果没有，那么有些任务会负担很重，有些任务会很忙。这将导致严重的缓慢。请发送groupBy ID and count 来衡量这一点。
2. 如果连接键自然偏斜，则向连接条件添加更多列以保持结果相同。更多的列可能会增加统一打乱数据的机会。这一点很难实现。
3. 内存需求取决于 - 正在运行的并行任务的数量、在执行器中执行的每个任务的数据量。减少其中一个或两个都将减少内存压力，并且显然运行速度较慢，但​​这比崩溃要好。我会通过在数据上创建更多分区来减少每个任务的数据量。假设您有 10 个分区用于 1B 行，然后将其设为 200 以减少每个任务的卷。在表 A 上使用repartition。不要创建太多分区，因为这会导致效率低下，10K 分区可能是个坏主意。
4. 有一些参数需要调整，即explained here。
5. 具有 10K 行的小表应该自动广播，因为它很小。如果没有，您可以增加广播限制并应用广播提示。