Skip to content

[SPARK-16221][SQL] Redirect Parquet JUL logger via SLF4J for WRITE operations #13918

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
dongjoon-hyun wants to merge 3 commits into from
Closed

[SPARK-16221][SQL] Redirect Parquet JUL logger via SLF4J for WRITE operations #13918

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a643114
[SPARK-16221][SQL] Redirect Parquet JUL logger via SLF4J for WRITE op…
dongjoon-hyun Jun 27, 2016
badf828
Revert the change about SPARK-9974.
dongjoon-hyun Jun 27, 2016
fd84576
Add ().
dongjoon-hyun Jun 27, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 12 additions & 5 deletions ...src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,8 @@ private[sql] class ParquetFileFormat
conf.setBoolean(ParquetOutputFormat.ENABLE_JOB_SUMMARY, false)
}

ParquetFileFormat.redirectParquetLogs()

new OutputWriterFactory {
override def newInstance(
path: String,
Expand Down Expand Up @@ -468,9 +470,9 @@ private[sql] class ParquetOutputWriterFactory(
override private[sql] def newWriter(path: String): OutputWriter = new OutputWriter {

// Create TaskAttemptContext that is used to pass on Configuration to the ParquetRecordWriter
private val hadoopTaskAttempId = new TaskAttemptID(new TaskID(new JobID, TaskType.MAP, 0), 0)
private val hadoopTaskAttemptId = new TaskAttemptID(new TaskID(new JobID, TaskType.MAP, 0), 0)
private val hadoopAttemptContext = new TaskAttemptContextImpl(
serializableConf.value, hadoopTaskAttempId)
serializableConf.value, hadoopTaskAttemptId)

// Instance of ParquetRecordWriter that does not use OutputCommitter
private val recordWriter = createNoCommitterRecordWriter(path, hadoopAttemptContext)
Expand Down Expand Up @@ -505,7 +507,7 @@ private[sql] class ParquetOutputWriterFactory(
dataSchema: StructType,
context: TaskAttemptContext): OutputWriter = {
throw new UnsupportedOperationException(
"this verison of newInstance not supported for " +
"this version of newInstance not supported for " +
"ParquetOutputWriterFactory")
}
}
Expand Down Expand Up @@ -665,7 +667,7 @@ private[sql] object ParquetFileFormat extends Logging {
Some(Try(DataType.fromJson(serializedSchema.get))
.recover { case _: Throwable =>
logInfo(
s"Serialized Spark schema in Parquet key-value metadata is not in JSON format, " +
"Serialized Spark schema in Parquet key-value metadata is not in JSON format, " +
"falling back to the deprecated DataType.fromCaseClassString parser.")
LegacyTypeStringParser.parse(serializedSchema.get)
}
Expand Down Expand Up @@ -880,7 +882,7 @@ private[sql] object ParquetFileFormat extends Logging {
Try(DataType.fromJson(schemaString).asInstanceOf[StructType]).recover {
case _: Throwable =>
logInfo(
s"Serialized Spark schema in Parquet key-value metadata is not in JSON format, " +
"Serialized Spark schema in Parquet key-value metadata is not in JSON format, " +
"falling back to the deprecated DataType.fromCaseClassString parser.")
LegacyTypeStringParser.parse(schemaString).asInstanceOf[StructType]
}.recoverWith {
Expand Down Expand Up @@ -926,4 +928,9 @@ private[sql] object ParquetFileFormat extends Logging {
// should be removed after this issue is fixed.
}
}

/**
* ParquetFileFormat.prepareWrite calls this function to initialize `redirectParquetLogsViaSLF4J`.
*/
def redirectParquetLogs(): Unit = {}
}