Zmień przedrostek nazwy pliku wyjściowego dla DataFrame.write()

Question

Pliki wyjściowe wygenerowane za pomocą metody Spark SQL DataFrame.write() rozpoczynają się od prefiksu "część". na przykład

DataFrame sample_07 = hiveContext.table("sample_07"); 
sample_07.write().parquet("sample_07_parquet"); 

Wyniki w:

hdfs dfs -ls sample_07_parquet/                                        
Found 4 items 
-rw-r--r-- 1 rob rob   0 2016-03-19 16:40 sample_07_parquet/_SUCCESS 
-rw-r--r-- 1 rob rob  491 2016-03-19 16:40 sample_07_parquet/_common_metadata 
-rw-r--r-- 1 rob rob  1025 2016-03-19 16:40 sample_07_parquet/_metadata 
-rw-r--r-- 1 rob rob  17194 2016-03-19 16:40 sample_07_parquet/part-r-00000-cefb2ac6-9f44-4ce4-93d9-8e7de3f2cb92.gz.parquet 

chciałbym zmienić przedrostek nazwy pliku wyjściowego używany podczas tworzenia pliku przy użyciu Spark SQL DataFrame.write(). Próbowałem ustawić właściwość "mapreduce.output.basename" w konfiguracji hadoop dla kontekstu Spark. na przykład

public class MyJavaSparkSQL { 

    public static void main(String[] args) throws Exception { 
    SparkConf sparkConf = new SparkConf().setAppName("MyJavaSparkSQL"); 
    JavaSparkContext ctx = new JavaSparkContext(sparkConf); 
    ctx.hadoopConfiguration().set("mapreduce.output.basename", "myprefix"); 
    HiveContext hiveContext = new org.apache.spark.sql.hive.HiveContext(ctx.sc()); 
    DataFrame sample_07 = hiveContext.table("sample_07"); 
    sample_07.write().parquet("sample_07_parquet"); 
    ctx.stop(); 
    } 

To nie zmieniło wyjściowego przedrostka nazwy pliku dla wygenerowanych plików.

Czy istnieje sposób na przesłonięcie przedrostka wyjściowej nazwy pliku podczas korzystania z metody DataFrame.write()?

Answer 1

Nie można zmienić prefiksu "części" podczas korzystania ze standardowych formatów wyjściowych (takich jak parkiet). Zobacz ten fragment z ParquetRelation source code:

private val recordWriter: RecordWriter[Void, InternalRow] = { 
    val outputFormat = { 
    new ParquetOutputFormat[InternalRow]() { 
     // ... 
     override def getDefaultWorkFile(context: TaskAttemptContext, extension: String): Path = { 
     // .. 
     // prefix is hard-coded here: 
     new Path(path, f"part-r-$split%05d-$uniqueWriteJobId$bucketString$extension") 
    } 
    } 
} 

Jeśli naprawdę musi kontrolować nazwy pliku części, prawdopodobnie będziesz musiał zaimplementować niestandardową FileOutputFormat i użyć jednego z Sparka zapisać metod, które akceptują klasę FileOutputFormat (np saveAsHadoopFile) .