Pyspark 写入 minio (s3) 分区失败答案

【问题标题】：Pyspark writing to minio (s3) partitioned failsPyspark 写入 minio (s3) 分区失败
【发布时间】：2021-12-28 14:33:00
【问题描述】：

我正在使用 pyspark 3.1.2 将文件写入 Minio S3。我正在使用分区，因此数据应存储在 batch_id 中，例如：

s3a://0001/transactions/batch_id=1 s3a://0001/transactions/batch_id=2 等等

写入本地文件系统时一切正常。

但是，当我将 S3 与分区提交程序一起使用时 (https://hadoop.apache.org/docs/r3.1.1/hadoop-aws/tools/hadoop-aws/committers.html)

带选项：“partitionOverwriteMode”=“静态” 例如： data_frame.write.mode("overwrite").partitionBy("batch_id").orc(output_path)

正在覆盖包括“transactions”在内的整个路径（而不是仅覆盖给定的分区）。

设置：

        spark_session.sparkContext._jsc.hadoopConfiguration().set(
            "fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem"
        )
        spark_session.sparkContext._jsc.hadoopConfiguration().set(
            "fs.s3a.path.style.access", "true"
        )
        spark_session.sparkContext._jsc.hadoopConfiguration().set(
            "fs.s3a.committer.magic.enabled", "true"
        )
        spark_session.sparkContext._jsc.hadoopConfiguration().set(
            "fs.s3a.committer.name", "partitioned"
        )
        spark_session.sparkContext._jsc.hadoopConfiguration().set(
            "fs.s3a.committer.staging.conflict-mode", "replace"
        )
        spark_session.sparkContext._jsc.hadoopConfiguration().set(
            "fs.s3a.committer.staging.abort.pending.uploads", "true"
        )

【问题讨论】：

更新：使用追加模式保存时，给定的分区不会被替换，而是递增。

标签： python amazon-s3 pyspark bigdata minio

【解决方案1】：

所以我添加了更多的罐子：

spark-hadoop-cloud_2.13-3.2.0.jar

并遵循 spark 云集成指南：[(https://spark.apache.org/docs/latest/cloud-integration.html)][1]

归结为添加：

"spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2"
"spark.sql.sources.commitProtocolClass", "org.apache.spark.internal.io.cloud.PathOutputCommitProtocol"
"spark.sql.parquet.output.committer.class", "org.apache.spark.internal.io.cloud.BindingParquetOutputCommitter"

然后切换回镶木地板。现在我可以在不覆盖孔路径的情况下覆盖分区。

【讨论】：