优化 Delta Lake 中的合并（Databricks 开源）答案

【问题标题】：Optimizing Merge in Delta Lake (Databricks Open Source )优化 Delta Lake 中的合并（Databricks 开源）
【发布时间】：2020-11-17 10:36:30
【问题描述】：

我正在尝试使用 delta Lake oss 实现合并，我的历史数据大约有 70 亿条记录，而 delta 大约有 500 万条记录。

合并基于复合键（5 列）。

我正在启动一个 10 节点集群 r5d.12xlarge（~3TB 内存/~480 个核心）。

该作业第一次需要 35 分钟，后续运行需要更多时间。

尝试使用优化技术，但没有任何效果，并且在 3 次运行后我开始出现堆内存问题，我在数据洗牌时看到磁盘上有很多溢出，尝试使用合并键上的 order by 重写历史记录，获得了性能改进和合并在 20 分钟内完成，泄漏量约为 2TB，但问题是作为合并过程的一部分写入的数据顺序不同，因为我无法控制写入数据的顺序，因此后续运行需要更长的时间。

我无法在 delta Lake oss 中使用 Zorder，因为它只附带订阅。我尝试了压缩，但这也没有帮助。如果有更好的方法来优化合并过程，请告诉我。

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql query-optimization databricks

【解决方案1】：

这是一个建议，您似乎正在 AWS 上运行您的 databricks 笔记本。

优化它以同时使用 Hive Metastore 或任何目录服务的方法。现在这有什么帮助？

在保存数据时，您可以使用bucketing 根据合并键对数据进行排序，并且此元数据信息需要存储在需要配置单元的元存储中。

如果您使用分桶，数据将井然有序，不会导致数据过度打乱，这将不可避免地提高您的工作性能。

我不太确定数据块，但如果您使用 EMR，您可以选择使用胶合目录作为元存储，或者您也可以在 EMR 中拥有自己的元存储。

【讨论】：

您好，感谢您的推荐。我正在尝试从 s3 读取和写入数据。看起来在使用 EMR 和 delta Lake 时，内存和磁盘中存在大量数据混洗和溢出。不确定，由于我对增量表分区没有任何控制权，因此如何减少洗牌和溢出。
将数据保留在 s3 本身中，但保留其元数据信息并使用 cluster by 来保存数据，这将使您的数据更少洗牌
是否可以使用 delta Lake 格式进行分桶？

【解决方案2】：

根据我的经验，20 分钟听起来不错；）您的分区方案是什么？合并的速度与 SELECTS 相同，因此如果您可以通过分区过滤器消除湖扫描，那将有很大帮助。

还可以查看 spark 中的随机分区设置，因为我发现这些设置会对性能产生巨大影响。

最后，压缩数据将对合并性能产生巨大影响。

【讨论】：

【解决方案3】：

如果你真的想通过代码来优化它，你可以启动并行任务。这是我们用来并行化 S3 编写的示例代码。您也可以对 adls 位置使用相同的逻辑。

with futures.ThreadPoolExecutor(max_workers=total_days+1) as e:
  print(f"{raw_bucket}/{db}/{table}/")
  for single_date in daterange(start_date, end_date):
    curr_date = single_date.strftime("%Y-%m-%d")
    jobs.append(e.submit(writeS3, curr_date))

  for job in futures.as_completed(jobs):
    result_done = job.result()
    print(f"Job Completed - {result_done}")

print("Task complete")

参考：https://docs.python.org/3/library/concurrent.futures.html

【讨论】：

跟Delta MERGE的问题有什么关系？

【解决方案4】：

我有同样的问题，同样的数据大小。我要去掉merge语句，把它分成两部分。

加入 where match，所以 UPSATE 语句。
在没有匹配的地方加入，插入。

【讨论】：