HeartSaVioR commented on a change in pull request #24128: [SPARK-27188][SS]
FileStreamSink: provide a new option to disable metadata log
URL: https://github.com/apache/spark/pull/24128#discussion_r266375643
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala
##########
@@ -125,16 +139,44 @@ class FileStreamSink(
partitionColumnNames: Seq[String],
options: Map[String, String]) extends Sink with Logging {
+ import FileStreamSink._
+
private val hadoopConf = sparkSession.sessionState.newHadoopConf()
private val basePath = new Path(path)
- private val logPath = {
- val metadataDir = new Path(basePath, FileStreamSink.metadataDir)
- val fs = metadataDir.getFileSystem(hadoopConf)
- FileStreamSink.checkEscapedMetadataPath(fs, metadataDir,
sparkSession.sessionState.conf)
- metadataDir
+
+ private val disableMetadata: Boolean = options.getOrElse("disableMetadata",
"false").toBoolean
+
+ private val fileLog: MetadataLog[Array[SinkFileStatus]] = {
+ if (disableMetadata) {
+ if (hasMetadata(Seq(path), hadoopConf, sparkSession.sessionState.conf)) {
+ throw new IllegalStateException("Metadata directory should not be
exist " +
+ "if 'disableMetadata' option is enabled.")
+ }
+ NoOpMetadataLog
+ } else {
+ val logPath = {
+ val metadataDir = new Path(basePath, FileStreamSink.metadataDir)
+ val fs = metadataDir.getFileSystem(hadoopConf)
+ FileStreamSink.checkEscapedMetadataPath(fs, metadataDir,
sparkSession.sessionState.conf)
+ metadataDir
+ }
+
+ new FileStreamSinkLog(FileStreamSinkLog.VERSION, sparkSession,
logPath.toString)
+ }
+ }
+
+ private val fileCommitProtocolClass = {
+ if (disableMetadata) {
+ // if metadata is disabled, just pick up same file commit protocol as
batch query
Review comment:
Would we want to just directly use SQLHadoopMapReduceCommitProtocol, or rely
on batch query file commit protocol as I'm proposing here?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
