Using Parquet writer API to do compatibility tests

Author: liancheng

sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetAvroCompatibilitySuite.scala

@@ -22,11 +22,9 @@ import java.util.{List => JList, Map => JMap}
 
 import scala.collection.JavaConversions._
 
-import org.apache.hadoop.mapreduce.Job
-import org.apache.parquet.avro.{AvroParquetOutputFormat, AvroWriteSupport}
-import org.apache.parquet.hadoop.ParquetOutputFormat
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.avro.AvroParquetWriter
 
-import org.apache.spark.rdd.RDD._
 import org.apache.spark.sql.parquet.test.avro.{Nested, ParquetAvroCompat}
 import org.apache.spark.sql.test.TestSQLContext
 import org.apache.spark.sql.{Row, SQLContext}
@@ -88,21 +86,13 @@ class ParquetAvroCompatibilitySuite extends ParquetCompatibilityTest {
   override protected def beforeAll(): Unit = {
     super.beforeAll()
 
-    val job = new Job()
-    ParquetOutputFormat.setWriteSupportClass(job, classOf[AvroWriteSupport])
-    AvroParquetOutputFormat.setSchema(job, ParquetAvroCompat.getClassSchema)
-
-    sqlContext
-      .sparkContext
-      .parallelize(0 until 10)
-      .map(i => (null, makeParquetAvroCompat(i)))
-      .coalesce(1)
-      .saveAsNewAPIHadoopFile(
-        parquetStore.getCanonicalPath,
-        classOf[Void],
-        classOf[ParquetAvroCompat],
-        classOf[ParquetOutputFormat[ParquetAvroCompat]],
-        job.getConfiguration)
+    val writer =
+      new AvroParquetWriter[ParquetAvroCompat](
+        new Path(parquetStore.getCanonicalPath),
+        ParquetAvroCompat.getClassSchema)
+
+    (0 until 10).foreach(i => writer.write(makeParquetAvroCompat(i)))
+    writer.close()
   }
 
   test("Read Parquet file generated by parquet-avro") {

sql/core/src/test/scala/org/apache/spark/sql/parquet/ParquetThriftCompatibilitySuite.scala

@@ -22,9 +22,9 @@ import java.util.{List => JList, Map => JMap}
 
 import scala.collection.JavaConversions._
 
-import org.apache.hadoop.mapreduce.Job
-import org.apache.parquet.hadoop.ParquetOutputFormat
-import org.apache.parquet.hadoop.thrift.ParquetThriftOutputFormat
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.hadoop.metadata.CompressionCodecName
+import org.apache.parquet.thrift.ThriftParquetWriter
 
 import org.apache.spark.sql.parquet.test.thrift.{Nested, ParquetThriftCompat, Suit}
 import org.apache.spark.sql.test.TestSQLContext
@@ -84,21 +84,14 @@ class ParquetThriftCompatibilitySuite extends ParquetCompatibilityTest {
   override protected def beforeAll(): Unit = {
     super.beforeAll()
 
-    val job = new Job()
-    ParquetThriftOutputFormat.setThriftClass(job, classOf[ParquetThriftCompat])
-    ParquetOutputFormat.setWriteSupportClass(job, classOf[ParquetThriftCompat])
-
-    sqlContext
-      .sparkContext
-      .parallelize(0 until 10)
-      .map(i => (null, makeParquetThriftCompat(i)))
-      .coalesce(1)
-      .saveAsNewAPIHadoopFile(
-        parquetStore.getCanonicalPath,
-        classOf[Void],
+    val writer =
+      new ThriftParquetWriter[ParquetThriftCompat](
+        new Path(parquetStore.getCanonicalPath),
         classOf[ParquetThriftCompat],
-        classOf[ParquetThriftOutputFormat[ParquetThriftCompat]],
-        job.getConfiguration)
+        CompressionCodecName.SNAPPY)
+
+    (0 until 10).foreach(i => writer.write(makeParquetThriftCompat(i)))
+    writer.close()
   }
 
   test("Read Parquet file generated by parquet-thrift") {