更改DataFrame.write()的输出文件名前缀
通过Spark SQL DataFrame.write()方法生成的输出文件以“part”basename前缀开头。 例如
DataFrame sample_07 = hiveContext.table("sample_07"); sample_07.write().parquet("sample_07_parquet");
结果是:
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
我想更改使用Spark SQL DataFrame.write()创建文件时使用的输出文件名前缀。 我尝试在Spark上下文的hadoop配置中设置“mapreduce.output.basename”属性。 例如
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(); }
这没有更改生成的文件的输出文件名前缀。
有没有办法在使用DataFrame.write()方法时覆盖输出文件名前缀?
使用任何标准输出格式(如Parquet)时,无法更改“part”前缀。 从ParquetRelation 源代码中查看此代码段:
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") } } }
如果你真的必须控制零件文件名,你可能必须实现一个自定义FileOutputFormat并使用一个接受FileOutputFormat类的Spark的保存方法(例如saveAsHadoopFile )。
- 将JavaPairRDD转换为JavaRDD
- Spark 2.0.1写入错误:引起:java.util.NoSuchElementException
- 带有DataFrame API的Apache Spark MLlib在createDataFrame()或read()时会产生java.net.URISyntaxException .csv(…)
- 如何在GroupBy操作后从spark DataFrame列中收集字符串列表?
- 在封闭范围内定义的局部变量日志必须是最终的或有效的最终
- apache spark MLLib:如何为字符串function构建标记点?
- 并行读取S3中的多个文件(Spark,Java)
- 使用RabbitMQ源的Spark结构化流式传输
- Spark – 任务不可序列化:如何使用调用外部类/对象的复杂映射闭包?