senyoung created SPARK-27526:
--------------------------------
Summary: Driver OOM error occurs while writing parquet file with
Append mode
Key: SPARK-27526
URL: https://issues.apache.org/jira/browse/SPARK-27526
Project: Spark
Issue Type: Bug
Components: Input/Output, SQL
Affects Versions: 2.1.1
Environment: centos6.7
Reporter: senyoung
As this user code below
{code:java}
someDataFrame.write
.mode(SaveMode.Append)
.partitionBy(somePartitionKeySeqs)
.parquet(targetPath);
{code}
When spark try to write parquet files into hdfs with the SaveMode.Append
mode,it must check the existing Partition Columns
would match the "existed files" ,how ever,this behevior would cache all leaf
fileInfos under the "targetPath";
This can easily trigger oom when there are too many files in the targetPath;
This behevior is useful when someone needs the exactly correctness ,but i
think it should be optional to avoid the oom;
The linked code be here
{code:java}
//package org.apache.spark.sql.execution.datasources
//case class DataSource
private def writeInFileFormat(format: FileFormat, mode: SaveMode, data:
DataFrame): Unit = {
...
if (mode == SaveMode.Append) {//can we make it optional?
val existingPartitionColumns = Try {
/**
* getOrInferFileFormatSchema(format, justPartitioning = true),
* this method may cause oom when there be too many files,could we just sample
limited files
* rather than all existed files ?
*/
getOrInferFileFormatSchema(format, justPartitioning = true)
._2.fieldNames.toList
}.getOrElse(Seq.empty[String])
// TODO: Case sensitivity.
val sameColumns =
existingPartitionColumns.map(_.toLowerCase()) ==
partitionColumns.map(_.toLowerCase())
if (existingPartitionColumns.nonEmpty && !sameColumns) {
throw new AnalysisException(
s"""Requested partitioning does not match existing partitioning.
|Existing partitioning columns:
| ${existingPartitionColumns.mkString(", ")}
|Requested partitioning columns:
| ${partitionColumns.mkString(", ")}
|""".stripMargin)
}
}
...
}
private def getOrInferFileFormatSchema(
format: FileFormat,
justPartitioning: Boolean = false): (StructType, StructType) = {
// the operations below are expensive therefore try not to do them if we
don't need to, e.g.,
// in streaming mode, we have already inferred and registered partition
columns, we will
// never have to materialize the lazy val below
lazy val tempFileIndex = {
val allPaths = caseInsensitiveOptions.get("path") ++ paths
val hadoopConf = sparkSession.sessionState.newHadoopConf()
val globbedPaths = allPaths.toSeq.flatMap { path =>
val hdfsPath = new Path(path)
val fs = hdfsPath.getFileSystem(hadoopConf)
val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory)
SparkHadoopUtil.get.globPathIfNecessary(qualified)
}.toArray
// InMemoryFileIndex.refresh0() cache all files info ,oom risks
new InMemoryFileIndex(sparkSession, globbedPaths, options, None)
}
val partitionSchema = if (partitionColumns.isEmpty) {
// Try to infer partitioning, because no DataSource in the read path
provides the partitioning
// columns properly unless it is a Hive DataSource
val resolved = tempFileIndex.partitionSchema.map { partitionField =>
val equality = sparkSession.sessionState.conf.resolver
// SPARK-18510: try to get schema from userSpecifiedSchema, otherwise
fallback to inferred
userSpecifiedSchema.flatMap(_.find(f => equality(f.name,
partitionField.name))).getOrElse(
partitionField)
}
StructType(resolved)
} else {
// maintain old behavior before SPARK-18510. If userSpecifiedSchema is
empty used inferred
// partitioning
if (userSpecifiedSchema.isEmpty) {
val inferredPartitions = tempFileIndex.partitionSchema
inferredPartitions
} else {
val partitionFields = partitionColumns.map { partitionColumn =>
val equality = sparkSession.sessionState.conf.resolver
userSpecifiedSchema.flatMap(_.find(c => equality(c.name,
partitionColumn))).orElse {
val inferredPartitions = tempFileIndex.partitionSchema
val inferredOpt = inferredPartitions.find(p => equality(p.name,
partitionColumn))
if (inferredOpt.isDefined) {
logDebug(
s"""Type of partition column: $partitionColumn not found in
specified schema
|for $format.
|User Specified Schema
|=====================
|${userSpecifiedSchema.orNull}
|
|Falling back to inferred dataType if it exists.
""".stripMargin)
}
inferredOpt
}.getOrElse {
throw new AnalysisException(s"Failed to resolve the schema for
$format for " +
s"the partition column: $partitionColumn. It must be specified
manually.")
}
}
StructType(partitionFields)
}
}
if (justPartitioning) {
return (null, partitionSchema)
}
val dataSchema = userSpecifiedSchema.map { schema =>
val equality = sparkSession.sessionState.conf.resolver
StructType(schema.filterNot(f => partitionSchema.exists(p =>
equality(p.name, f.name))))
}.orElse {
format.inferSchema(
sparkSession,
caseInsensitiveOptions,
tempFileIndex.allFiles())
}.getOrElse {
throw new AnalysisException(
s"Unable to infer schema for $format. It must be specified manually.")
}
(dataSchema, partitionSchema)
}
{code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
