如何在 pyspark 中设置 spark.sql.parquet.output.committer.class

Question

我正在尝试设置spark.sql.parquet.output.committer.class，但我似乎没有做任何事情来使设置生效。

我试图让许多线程写入同一个输出文件夹，这将与 org.apache.spark.sql. parquet.DirectParquetOutputCommitter 一起使用，因为它不会使用 _temporary 文件夹。我收到以下错误，这就是我知道它不起作用的原因：

Caused by: java.io.FileNotFoundException: File hdfs://path/to/stuff/_temporary/0/task_201606281757_0048_m_000029/some_dir does not exist.
        at org.apache.hadoop.hdfs.DistributedFileSystem.listStatusInternal(DistributedFileSystem.java:795)
        at org.apache.hadoop.hdfs.DistributedFileSystem.access$700(DistributedFileSystem.java:106)
        at org.apache.hadoop.hdfs.DistributedFileSystem$18.doCall(DistributedFileSystem.java:853)
        at org.apache.hadoop.hdfs.DistributedFileSystem$18.doCall(DistributedFileSystem.java:849)
        at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
        at org.apache.hadoop.hdfs.DistributedFileSystem.listStatus(DistributedFileSystem.java:849)
        at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.mergePaths(FileOutputCommitter.java:382)
        at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.mergePaths(FileOutputCommitter.java:384)
        at org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter.commitJob(FileOutputCommitter.java:326)
        at org.apache.parquet.hadoop.ParquetOutputCommitter.commitJob(ParquetOutputCommitter.java:46)
        at org.apache.spark.sql.execution.datasources.BaseWriterContainer.commitJob(WriterContainer.scala:230)
        at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1.apply$mcV$sp(InsertIntoHadoopFsRelation.scala:151)

注意对默认类org.apache.parquet.hadoop.ParquetOutputCommitter.commitJob 的调用。

根据其他 SO 答案和搜索，我尝试了以下方法：

1. sc._jsc.hadoopConfiguration().set(key, val)（这确实适用于 parquet.enable.summary-metadata 等设置）
2. dataframe.write.option(key, val).parquet
3. 将--conf "spark.hadoop.spark.sql.parquet.output.committer.class=org.apache.spark.sql.parquet.DirectParquetOutputCommitter" 添加到spark-submit 调用中
4. 将--conf "spark.sql.parquet.output.committer.class"=" org.apache.spark.sql.parquet.DirectParquetOutputCommitter" 添加到spark-submit 调用中。

这就是我能找到的所有东西，但没有任何效果。 set in Scala 看起来并不难，但在 Python 中似乎是不可能的。

Answer 1

this comment 中的方法确实对我有用：

16/06/28 18:49:59 INFO ParquetRelation: Using user defined output committer for Parquet: org.apache.spark.sql.execution.datasources.parquet.DirectParquetOutputCommitter

这是 Spark 给出的洪水中丢失的日志消息，与我看到的错误无关。无论如何，这一切都没有实际意义，因为 DirectParquetOutputCommitter 一直是 removed from Spark。