python - 如何在 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-提交调用
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 .