在数据块上将严重分区的数据帧加速到 s3

Posted 2023-04-15

【中文标题】在数据块上将严重分区的数据帧加速到 s3

【英文标题】：speeding up heavily partitioned dataframe to s3 on databricks

【发布时间】：2020-08-24 14:56:21

【问题描述】：

我正在 Databricks 上运行一个笔记本，它创建分区的 PySpark 数据帧并将它们上传到 s3。有问题的表有 ~5,000 个文件，总大小约为 5 GB（需要以这种方式分区才能被 Athena 有效查询）。我的问题是，将文件写入 s3 似乎是顺序的而不是并行的，并且可能需要长达一小时。例如：

df.repartition("customer_id")
  .write.partitionBy("customer_id")
  .mode("overwrite")
  .format("parquet")
  .save("s3a://mybucket/path-to-table/")

我已经在 AWS 上使用以下配置启动了我的集群 (i3.xlarge)：

spark.hadoop.orc.overwrite.output.file true
spark.databricks.io.directoryCommit.enableLogicalDelete true
spark.sql.sources.commitProtocolClass org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol
parquet.enable.summary-metadata false
spark.hadoop.fs.s3.maxRetries 20
spark.databricks.hive.metastore.glueCatalog.enabled true
spark.hadoop.validateOutputSpecs false
mapreduce.fileoutputcommitter.marksuccessfuljobs false
spark.sql.legacy.parquet.datetimeRebaseModeInRead CORRECTED
spark.hadoop.fs.s3.consistent.retryPeriodSeconds 10
spark.speculation true
spark.hadoop.fs.s3.consistent true
spark.hadoop.fs.s3.consistent.retryCount 5

在我有很多小文件需要快速写入 s3 的情况下，推荐的方法是什么？

【问题讨论】：

为什么你的数据集需要顺序写入而不是并行写入？性能来自并行（写入）操作。你有多少个集群节点？

【参考方案1】：

我看到了你写速度慢的几个原因，可以加快速度：

您可能有超过 5,000 个客户？因此，使用分区依据，您可能有超过 5,000 个分区。由于元存储中的开销，使用 Parquet（非 Delta Lake 表）可能会非常慢。我不认为你想要这么多分区。 5GB 有 5,000 个文件，每个文件大小约为 1MB。这是非常小的。对于此问题，您写出的文件大小应接近 100MB。 默认集群选项设计得非常好，我很少需要更改它们，当我这样做时，我正在启用新功能。您应该尝试解决上述问题，并删除设置中的所有这些覆盖。 Repartition("customer_id") 和 partitionBy("customer_id") 是多余的。

推荐：

如果您的前一阶段创建了 > 50 个分区，则可以使用 coalesce() 将文件大小提高到 ~ 100MB。 通过 customer_id 删除分区，也许您认为这样做有充分的理由，但小文件和大量分区正在扼杀您的性能。 尝试开放的 Delta Lake 格式（例如 CREATE TABLE ... USING DELTA LOCATION ...。这将加快您的客户选择性查询，如果您还 OPTIMIZE ... ZORDER BY customer_id 将加快加入 customer_id 的速度，并且可以自动优化文件的大小。

最终结果看起来更清晰：

df.coalesce(50)
  .write
  .mode("overwrite")
  .format("delta")
  .save("s3a://mybucket/path-to-table/")

查看自动优化选项以自动调整文件大小：https://docs.databricks.com/delta/optimizations/auto-optimize.html#usage

Delta Lake 表可以与 Athena 一起使用 https://docs.databricks.com/delta/presto-integration.html#presto-and-athena-to-delta-lake-integration

【参考方案2】：

在s3桶上，你设置fs.s3a.fast.upload = true了吗？我正在link 上查看类似的票证