【发布时间】：2022-01-17 20:33:01
【问题描述】：
我目前正在使用 Apache Spark (pyspark) 构建应用程序，我有以下用例：
- 以本地模式运行
pyspark（使用spark-submit local[*]）。 - 将 Spark 作业的结果以分区 Parquet 文件的形式写入 S3。
- 确保每个作业都覆盖它正在写入的特定分区，以确保作业具有幂等性。
- 确保在提交到 S3 之前将 spark-staging 文件写入本地磁盘，因为在 S3 中暂存，然后通过重命名操作提交，成本非常高。
由于各种内部原因，上述所有四个要点都是不可协商的。
除了最后一个要点外，我什么都有。我正在运行一个pyspark 应用程序，并写入S3（实际上是一个本地Ceph 实例），确保spark.sql.sources.partitionOverwriteMode 设置为dynamic。
但是，这意味着我的 spark-staging 文件正在 S3 中暂存，然后使用删除和重命名操作提交，这非常昂贵。
我已尝试使用Spark Directory Committer 来暂存本地磁盘上的文件。这很好用，除非spark.sql.sources.partitionOverwriteMode。
After digging through the source code，看来 PathOutputCommitter 不支持动态分区覆盖。
在这一点上，我被困住了。我希望能够将我的暂存文件写入本地磁盘，然后将结果提交到 S3。但是，我还需要能够在不覆盖整个 Parquet 表的情况下动态覆盖单个分区。
作为参考，我正在运行 pyspark=3.1.2，并使用以下 spark-submit 命令：
spark-submit --repositories https://repository.cloudera.com/artifactory/cloudera-repos/ --packages com.amazonaws:aws-java-sdk:1.11.375,org.apache.hadoop:hadoop-aws:3.2.0,org.apache.spark:spark-hadoop-cloud_2.12:3.1.1.3.1.7270.0-253
当spark.sql.sources.partitionOverwriteMode 设置为dynamic 时出现以下错误：
java.io.IOException: PathOutputCommitProtocol does not support dynamicPartitionOverwrite
我的 spark 配置如下：
self.spark.conf.set("spark.sql.files.ignoreCorruptFiles", "true")
self.spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
self.spark.conf.set("spark.hadoop.fs.s3a.committer.name", "magic")
self.spark.conf.set("spark.sql.sources.commitProtocolClass",
"org.apache.spark.internal.io.cloud.PathOutputCommitProtocol")
self.spark.conf.set("spark.sql.parquet.output.committer.class",
"org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter")
self.spark.conf.set(
"spark.sql.sources.partitionOverwriteMode", "dynamic"
)
【问题讨论】：
标签： amazon-web-services apache-spark amazon-s3 pyspark parquet