From 0ef8caf461d0a0ef7eee6bbc78db600bc46dc846 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Wed, 1 Jul 2015 17:41:53 +0800 Subject: [PATCH 1/7] Keep cached information and avoid re-calculating footers. --- .../apache/spark/sql/parquet/newParquet.scala | 46 +++++++++++++------ 1 file changed, 31 insertions(+), 15 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala index bc39fae2bcfde..509c732184094 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala @@ -331,33 +331,49 @@ private[sql] class ParquetRelation2( // Schema of the whole table, including partition columns. var schema: StructType = _ + // Cached leaf statuses + var localCachedLeafStatuses: Set[FileStatus] = _ + + // Cached leaves + var cachedLeaves: Array[FileStatus] = _ + /** * Refreshes `FileStatus`es, footers, partition spec, and table schema. */ def refresh(): Unit = { + // Check if cachedLeafStatuses is changed or not + val leafStatusesChanged = localCachedLeafStatuses != cachedLeafStatuses() + // Lists `FileStatus`es of all leaf nodes (files) under all base directories. - val leaves = cachedLeafStatuses().filter { f => - isSummaryFile(f.getPath) || - !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith(".")) - }.toArray + val leaves = if (leafStatusesChanged) { + localCachedLeafStatuses = cachedLeafStatuses() + cachedLeaves = cachedLeafStatuses().filter { f => + isSummaryFile(f.getPath) || + !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith(".")) + }.toArray + cachedLeaves + } else { + cachedLeaves + } dataStatuses = leaves.filterNot(f => isSummaryFile(f.getPath)) metadataStatuses = leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_METADATA_FILE) commonMetadataStatuses = leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_COMMON_METADATA_FILE) - footers = { - val conf = SparkHadoopUtil.get.conf - val taskSideMetaData = conf.getBoolean(ParquetInputFormat.TASK_SIDE_METADATA, true) - val rawFooters = if (shouldMergeSchemas) { - ParquetFileReader.readAllFootersInParallel( - conf, seqAsJavaList(leaves), taskSideMetaData) - } else { - ParquetFileReader.readAllFootersInParallelUsingSummaryFiles( - conf, seqAsJavaList(leaves), taskSideMetaData) + if (leafStatusesChanged) { + footers = { + val conf = SparkHadoopUtil.get.conf + val taskSideMetaData = conf.getBoolean(ParquetInputFormat.TASK_SIDE_METADATA, true) + val rawFooters = if (shouldMergeSchemas) { + ParquetFileReader.readAllFootersInParallel( + conf, seqAsJavaList(leaves), taskSideMetaData) + } else { + ParquetFileReader.readAllFootersInParallelUsingSummaryFiles( + conf, seqAsJavaList(leaves), taskSideMetaData) + } + rawFooters.map(footer => footer.getFile -> footer).toMap } - - rawFooters.map(footer => footer.getFile -> footer).toMap } // If we already get the schema, don't need to re-compute it since the schema merging is From 12a0ed92a9ed310166b9fb50bc6b971fa18ad4ac Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Mon, 20 Jul 2015 14:36:32 +0800 Subject: [PATCH 2/7] Add check of FileStatus's modification time. --- .../apache/spark/sql/parquet/newParquet.scala | 35 ++++++++++--------- 1 file changed, 19 insertions(+), 16 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala index d3c6447fb75a8..93c0c647c04f1 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala @@ -349,7 +349,7 @@ private[sql] class ParquetRelation2( private var commonMetadataStatuses: Array[FileStatus] = _ // Parquet footer cache. - var footers: Map[Path, Footer] = _ + val footers: mutable.Map[Path, Footer] = mutable.Map[Path, Footer]() // `FileStatus` objects of all data files (Parquet part-files). var dataStatuses: Array[FileStatus] = _ @@ -364,8 +364,10 @@ private[sql] class ParquetRelation2( // Cached leaf statuses var localCachedLeafStatuses: Set[FileStatus] = _ + var lastRefreshTime: Long = 0 + // Cached leaves - var cachedLeaves: Array[FileStatus] = _ + var cachedLeaves: Array[FileStatus] = Array() /** * Refreshes `FileStatus`es, footers, partition spec, and table schema. @@ -377,10 +379,13 @@ private[sql] class ParquetRelation2( // Lists `FileStatus`es of all leaf nodes (files) under all base directories. val leaves = if (leafStatusesChanged) { localCachedLeafStatuses = cachedLeafStatuses() - cachedLeaves = cachedLeafStatuses().filter { f => - isSummaryFile(f.getPath) || - !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith(".")) + val updatedLeaves = cachedLeafStatuses().filter { f => + (isSummaryFile(f.getPath) || + !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith("."))) && + (!footers.contains(f.getPath) || f.getModificationTime > lastRefreshTime) }.toArray + lastRefreshTime = System.currentTimeMillis + cachedLeaves = updatedLeaves cachedLeaves } else { cachedLeaves @@ -392,18 +397,16 @@ private[sql] class ParquetRelation2( leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_COMMON_METADATA_FILE) if (leafStatusesChanged) { - footers = { - val conf = SparkHadoopUtil.get.conf - val taskSideMetaData = conf.getBoolean(ParquetInputFormat.TASK_SIDE_METADATA, true) - val rawFooters = if (shouldMergeSchemas) { - ParquetFileReader.readAllFootersInParallel( - conf, seqAsJavaList(leaves), taskSideMetaData) - } else { - ParquetFileReader.readAllFootersInParallelUsingSummaryFiles( - conf, seqAsJavaList(leaves), taskSideMetaData) - } - rawFooters.map(footer => footer.getFile -> footer).toMap + val conf = SparkHadoopUtil.get.conf + val taskSideMetaData = conf.getBoolean(ParquetInputFormat.TASK_SIDE_METADATA, true) + val rawFooters = if (shouldMergeSchemas) { + ParquetFileReader.readAllFootersInParallel( + conf, seqAsJavaList(leaves), taskSideMetaData) + } else { + ParquetFileReader.readAllFootersInParallelUsingSummaryFiles( + conf, seqAsJavaList(leaves), taskSideMetaData) } + rawFooters.map(footer => footers += (footer.getFile -> footer)) } // If we already get the schema, don't need to re-compute it since the schema merging is From fa5458f53c578630430c84faba284b05e3b96487 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Wed, 22 Jul 2015 16:35:32 +0800 Subject: [PATCH 3/7] Use Map to cache FileStatus and do merging previously loaded schema and newly loaded one. --- .../apache/spark/sql/parquet/newParquet.scala | 92 ++++++++++--------- 1 file changed, 48 insertions(+), 44 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala index b61f061950931..1c5ae7840771b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala @@ -345,59 +345,59 @@ private[sql] class ParquetRelation2( // Schema of the whole table, including partition columns. var schema: StructType = _ - // Cached leaf statuses - var localCachedLeafStatuses: Set[FileStatus] = _ - - var lastRefreshTime: Long = 0 - // Cached leaves - var cachedLeaves: Array[FileStatus] = Array() + val cachedLeaves: mutable.Map[Path, FileStatus] = new mutable.HashMap[Path, FileStatus]() /** * Refreshes `FileStatus`es, footers, partition spec, and table schema. */ def refresh(): Unit = { - // Check if cachedLeafStatuses is changed or not - val leafStatusesChanged = localCachedLeafStatuses != cachedLeafStatuses() - // Lists `FileStatus`es of all leaf nodes (files) under all base directories. - val leaves = if (leafStatusesChanged) { - localCachedLeafStatuses = cachedLeafStatuses() - val updatedLeaves = cachedLeafStatuses().filter { f => - (isSummaryFile(f.getPath) || - !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith("."))) && - (f.getModificationTime > lastRefreshTime) - }.toArray - lastRefreshTime = System.currentTimeMillis - cachedLeaves = updatedLeaves - cachedLeaves - } else { - cachedLeaves - } + // We only care about new FileStatus and updated FileStatus. + val leaves = cachedLeafStatuses().filter { f => + (isSummaryFile(f.getPath) || + !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith("."))) && + (!cachedLeaves.contains(f.getPath) || + cachedLeaves(f.getPath).getModificationTime < f.getModificationTime) + }.map { f => + cachedLeaves += (f.getPath -> f) + f + }.toArray dataStatuses = leaves.filterNot(f => isSummaryFile(f.getPath)) metadataStatuses = leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_METADATA_FILE) commonMetadataStatuses = leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_COMMON_METADATA_FILE) - // If we already get the schema, don't need to re-compute it since the schema merging is - // time-consuming. - if (dataSchema == null) { - dataSchema = { - val dataSchema0 = maybeDataSchema - .orElse(readSchema()) - .orElse(maybeMetastoreSchema) - .getOrElse(throw new AnalysisException( - s"Failed to discover schema of Parquet file(s) in the following location(s):\n" + - paths.mkString("\n\t"))) - - // If this Parquet relation is converted from a Hive Metastore table, must reconcile case - // case insensitivity issue and possible schema mismatch (probably caused by schema - // evolution). - maybeMetastoreSchema - .map(ParquetRelation2.mergeMetastoreParquetSchema(_, dataSchema0)) - .getOrElse(dataSchema0) - } + dataSchema = { + val dataSchema0 = maybeDataSchema + .orElse(readSchema()) + .orElse(maybeMetastoreSchema) + + // If we already have previous loaded schema, merge it with newly loaded schema. + val dataSchema1 = + if (dataSchema != null) { + if (dataSchema0.isDefined) { + dataSchema0.get.merge(dataSchema) + } else { + dataSchema + } + } else { + if (dataSchema0.isDefined) { + dataSchema0.get + } else { + throw new AnalysisException( + s"Failed to discover schema of Parquet file(s) in the following location(s):\n" + + paths.mkString("\n\t")) + } + } + + // If this Parquet relation is converted from a Hive Metastore table, must reconcile case + // case insensitivity issue and possible schema mismatch (probably caused by schema + // evolution). + maybeMetastoreSchema + .map(ParquetRelation2.mergeMetastoreParquetSchema(_, dataSchema1)) + .getOrElse(dataSchema1) } } @@ -446,11 +446,15 @@ private[sql] class ParquetRelation2( } assert( - filesToTouch.nonEmpty || maybeDataSchema.isDefined || maybeMetastoreSchema.isDefined, - "No predefined schema found, " + - s"and no Parquet data files or summary files found under ${paths.mkString(", ")}.") + filesToTouch.nonEmpty || maybeDataSchema.isDefined || maybeMetastoreSchema.isDefined || + dataSchema != null, "No predefined schema found, " + + s"and no Parquet data files or summary files found under ${paths.mkString(", ")}.") - ParquetRelation2.mergeSchemasInParallel(filesToTouch, sqlContext) + if (filesToTouch.nonEmpty) { + ParquetRelation2.mergeSchemasInParallel(filesToTouch, sqlContext) + } else { + None + } } } } From a52b6d1036d07353fd0c97d095b8b0019b2bbcc6 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Thu, 23 Jul 2015 15:32:01 +0800 Subject: [PATCH 4/7] For comments. --- .../apache/spark/sql/parquet/newParquet.scala | 91 ++++++++----------- 1 file changed, 38 insertions(+), 53 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala index 1c5ae7840771b..45a0e14749f6a 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala @@ -346,58 +346,47 @@ private[sql] class ParquetRelation2( var schema: StructType = _ // Cached leaves - val cachedLeaves: mutable.Map[Path, FileStatus] = new mutable.HashMap[Path, FileStatus]() + var cachedLeaves: Set[FileStatus] = Set() /** * Refreshes `FileStatus`es, footers, partition spec, and table schema. */ def refresh(): Unit = { - // Lists `FileStatus`es of all leaf nodes (files) under all base directories. - // We only care about new FileStatus and updated FileStatus. - val leaves = cachedLeafStatuses().filter { f => - (isSummaryFile(f.getPath) || - !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith("."))) && - (!cachedLeaves.contains(f.getPath) || - cachedLeaves(f.getPath).getModificationTime < f.getModificationTime) - }.map { f => - cachedLeaves += (f.getPath -> f) - f - }.toArray - - dataStatuses = leaves.filterNot(f => isSummaryFile(f.getPath)) - metadataStatuses = leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_METADATA_FILE) - commonMetadataStatuses = - leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_COMMON_METADATA_FILE) - - dataSchema = { - val dataSchema0 = maybeDataSchema - .orElse(readSchema()) - .orElse(maybeMetastoreSchema) - - // If we already have previous loaded schema, merge it with newly loaded schema. - val dataSchema1 = - if (dataSchema != null) { - if (dataSchema0.isDefined) { - dataSchema0.get.merge(dataSchema) - } else { - dataSchema - } - } else { - if (dataSchema0.isDefined) { - dataSchema0.get - } else { - throw new AnalysisException( - s"Failed to discover schema of Parquet file(s) in the following location(s):\n" + - paths.mkString("\n\t")) - } - } + val currentLeafStatuses = cachedLeafStatuses() + + // Check if cachedLeafStatuses is changed or not + val leafStatusesChanged = !cachedLeaves.equals(currentLeafStatuses) + + if (leafStatusesChanged) { + cachedLeaves = currentLeafStatuses.filter { f => + isSummaryFile(f.getPath) || + !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith(".")) + } - // If this Parquet relation is converted from a Hive Metastore table, must reconcile case - // case insensitivity issue and possible schema mismatch (probably caused by schema - // evolution). - maybeMetastoreSchema - .map(ParquetRelation2.mergeMetastoreParquetSchema(_, dataSchema1)) - .getOrElse(dataSchema1) + // Lists `FileStatus`es of all leaf nodes (files) under all base directories. + val leaves = cachedLeaves.toArray + + dataStatuses = leaves.filterNot(f => isSummaryFile(f.getPath)) + metadataStatuses = + leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_METADATA_FILE) + commonMetadataStatuses = + leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_COMMON_METADATA_FILE) + + dataSchema = { + val dataSchema0 = maybeDataSchema + .orElse(readSchema()) + .orElse(maybeMetastoreSchema) + .getOrElse(throw new AnalysisException( + s"Failed to discover schema of Parquet file(s) in the following location(s):\n" + + paths.mkString("\n\t"))) + + // If this Parquet relation is converted from a Hive Metastore table, must reconcile case + // case insensitivity issue and possible schema mismatch (probably caused by schema + // evolution). + maybeMetastoreSchema + .map(ParquetRelation2.mergeMetastoreParquetSchema(_, dataSchema0)) + .getOrElse(dataSchema0) + } } } @@ -446,15 +435,11 @@ private[sql] class ParquetRelation2( } assert( - filesToTouch.nonEmpty || maybeDataSchema.isDefined || maybeMetastoreSchema.isDefined || - dataSchema != null, "No predefined schema found, " + + filesToTouch.nonEmpty || maybeDataSchema.isDefined || maybeMetastoreSchema.isDefined, + "No predefined schema found, " + s"and no Parquet data files or summary files found under ${paths.mkString(", ")}.") - if (filesToTouch.nonEmpty) { - ParquetRelation2.mergeSchemasInParallel(filesToTouch, sqlContext) - } else { - None - } + ParquetRelation2.mergeSchemasInParallel(filesToTouch, sqlContext) } } } From c8fdfb7757ef0eadd6691a920d68bee136fc6ea7 Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Thu, 23 Jul 2015 17:39:02 +0800 Subject: [PATCH 5/7] Fix it. --- .../main/scala/org/apache/spark/sql/parquet/newParquet.scala | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala index 45a0e14749f6a..56eafbefc0fb3 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala @@ -346,7 +346,7 @@ private[sql] class ParquetRelation2( var schema: StructType = _ // Cached leaves - var cachedLeaves: Set[FileStatus] = Set() + var cachedLeaves: Set[FileStatus] = null /** * Refreshes `FileStatus`es, footers, partition spec, and table schema. @@ -355,7 +355,8 @@ private[sql] class ParquetRelation2( val currentLeafStatuses = cachedLeafStatuses() // Check if cachedLeafStatuses is changed or not - val leafStatusesChanged = !cachedLeaves.equals(currentLeafStatuses) + val leafStatusesChanged = (cachedLeaves == null) || + !cachedLeaves.equals(currentLeafStatuses) if (leafStatusesChanged) { cachedLeaves = currentLeafStatuses.filter { f => From ae0ec64b506114218f059592289aece72aaf56ae Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Thu, 23 Jul 2015 22:39:01 +0800 Subject: [PATCH 6/7] Fix wrong assignment. --- .../org/apache/spark/sql/parquet/newParquet.scala | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala index 56eafbefc0fb3..20adddbaaeebb 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala @@ -359,13 +359,13 @@ private[sql] class ParquetRelation2( !cachedLeaves.equals(currentLeafStatuses) if (leafStatusesChanged) { - cachedLeaves = currentLeafStatuses.filter { f => - isSummaryFile(f.getPath) || - !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith(".")) - } + cachedLeaves = currentLeafStatuses.toIterator.toSet // Lists `FileStatus`es of all leaf nodes (files) under all base directories. - val leaves = cachedLeaves.toArray + val leaves = currentLeafStatuses.filter { f => + isSummaryFile(f.getPath) || + !(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith(".")) + }.toArray dataStatuses = leaves.filterNot(f => isSummaryFile(f.getPath)) metadataStatuses = From 92e93472939edeeb29079a681c7c2bf3a6477f1e Mon Sep 17 00:00:00 2001 From: Liang-Chi Hsieh Date: Fri, 24 Jul 2015 17:06:21 +0800 Subject: [PATCH 7/7] Fix indentation. --- .../main/scala/org/apache/spark/sql/parquet/newParquet.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala index 20adddbaaeebb..c384697c0ee62 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala @@ -437,8 +437,8 @@ private[sql] class ParquetRelation2( assert( filesToTouch.nonEmpty || maybeDataSchema.isDefined || maybeMetastoreSchema.isDefined, - "No predefined schema found, " + - s"and no Parquet data files or summary files found under ${paths.mkString(", ")}.") + "No predefined schema found, " + + s"and no Parquet data files or summary files found under ${paths.mkString(", ")}.") ParquetRelation2.mergeSchemasInParallel(filesToTouch, sqlContext) }