Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/12409#discussion_r59942518
--- Diff:
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala
---
@@ -371,6 +382,97 @@ private[sql] class DefaultSource
}
}
+private[sql] class ParquetOutputWriterFactory(
+ sqlConf: SQLConf,
+ dataSchema: StructType,
+ hadoopConf: Configuration,
+ options: Map[String, String])
+ extends OutputWriterFactory {
+
+ @transient private val preparedHadoopConf = {
+
+ val job = Job.getInstance(hadoopConf)
+ val conf = ContextUtil.getConfiguration(job)
+ val parquetOptions = new ParquetOptions(options, sqlConf)
+
+ // We're not really using `ParquetOutputFormat[Row]` for writing data
here, because we override
+ // it in `ParquetOutputWriter` to support appending and dynamic
partitioning. The reason why
+ // we set it here is to setup the output committer class to
`ParquetOutputCommitter`, which is
+ // bundled with `ParquetOutputFormat[Row]`.
+ job.setOutputFormatClass(classOf[ParquetOutputFormat[Row]])
+
+ ParquetOutputFormat.setWriteSupportClass(job,
classOf[CatalystWriteSupport])
+
+ // We want to clear this temporary metadata from saving into Parquet
file.
+ // This metadata is only useful for detecting optional columns when
pushdowning filters.
+ val dataSchemaToWrite = StructType.removeMetadata(
+ StructType.metadataKeyForOptionalField,
+ dataSchema).asInstanceOf[StructType]
+ CatalystWriteSupport.setSchema(dataSchemaToWrite, conf)
+
+ // Sets flags for `CatalystSchemaConverter` (which converts Catalyst
schema to Parquet schema)
+ // and `CatalystWriteSupport` (writing actual rows to Parquet files).
+ conf.set(
+ SQLConf.PARQUET_BINARY_AS_STRING.key,
+ sqlConf.isParquetBinaryAsString.toString)
+
+ conf.set(
+ SQLConf.PARQUET_INT96_AS_TIMESTAMP.key,
+ sqlConf.isParquetINT96AsTimestamp.toString)
+
+ conf.set(
+ SQLConf.PARQUET_WRITE_LEGACY_FORMAT.key,
+ sqlConf.writeLegacyParquetFormat.toString)
+
+ // Sets compression scheme
+ conf.set(ParquetOutputFormat.COMPRESSION,
parquetOptions.compressionCodec)
+
+ conf
+ }
+
+ private val serializableConf = new
SerializableConfiguration(preparedHadoopConf)
--- End diff --
You know, I thought that was the case but then I found an example where
closure capture of the value for a broadcast slowed down the jobs by an order
of magnitude. So i'm not convinced this is safe. I guess we should either
verify or broadcast.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
