avoid to output empty parquet files

Author: uncleGen

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormatWriter.scala

@@ -292,7 +292,10 @@ object FileFormatWriter extends Logging {
     override def execute(iter: Iterator[InternalRow]): Set[String] = {
       var fileCounter = 0
       var recordsInFile: Long = 0L
-      newOutputWriter(fileCounter)
+      // Skip the empty partition to avoid creating a mass of 'empty' files.
+      if (iter.hasNext) {
+        newOutputWriter(fileCounter)
+      }
       while (iter.hasNext) {
         if (description.maxRecordsPerFile > 0 && recordsInFile >= description.maxRecordsPerFile) {
           fileCounter += 1

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala

@@ -140,10 +140,10 @@ class ParquetFileFormat
       // initialized.
       private val parquetLogRedirector = ParquetLogRedirector.INSTANCE
 
-        override def newInstance(
-          path: String,
-          dataSchema: StructType,
-          context: TaskAttemptContext): OutputWriter = {
+      override def newInstance(
+        path: String,
+        dataSchema: StructType,
+        context: TaskAttemptContext): OutputWriter = {
         new ParquetOutputWriter(path, context)
       }