Come impostare spark.sql.parquet.output.committer.class in pyspark

Question

Sto provando a impostare spark.sql.parquet.output.committer.class e non sembra che le impostazioni abbiano effetto.

Sto cercando di scrivere più thread sulla stessa cartella di output, che funzionerebbe con org.apache.spark.sql. parquet.DirectParquetOutputCommitter poiché non userebbe la cartella _temporary. Sto ottenendo il seguente errore, che è come so che non funziona:

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) 

nota la chiamata al org.apache.parquet.hadoop.ParquetOutputCommitter.commitJob, la classe predefinita.

Ho provato quanto segue, sulla base di altri SO risposte e ricerche:

1. sc._jsc.hadoopConfiguration().set(key, val) (questo funziona per le impostazioni come parquet.enable.summary-metadata)
2. dataframe.write.option(key, val).parquet
3. Aggiunta --conf "spark.hadoop.spark.sql.parquet.output.committer.class=org.apache.spark.sql.parquet.DirectParquetOutputCommitter" alla chiamata spark-submit
4. Aggiunta di --conf "spark.sql.parquet.output.committer.class"=" org.apache.spark.sql.parquet.DirectParquetOutputCommitter" alla chiamata spark-submit.

Questo è tutto ciò che sono riuscito a trovare e niente funziona. Sembra che non sia difficile da fare set in Scala ma sembra impossibile in Python.

Answer 1

L'approccio in this comment definitivamente funzionato per me:

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

E 'stato un messaggio di log perduto nel diluvio che scintilla dà, e l'errore che stavo vedendo era indipendente. È comunque tutto sommato, dal momento che DirectParquetOutputCommitter è stato removed from Spark.