Make ZSTD default compression for Parquet writes

scio-parquet/src/main/scala/com/spotify/scio/parquet/avro/ParquetAvroIO.scala

@@ -303,7 +303,7 @@ object ParquetAvroIO {
     private[scio] val DefaultSchema = null
     private[scio] val DefaultNumShards = 0
     private[scio] val DefaultSuffix = ".parquet"
-    private[scio] val DefaultCompression = CompressionCodecName.GZIP
+    private[scio] val DefaultCompression = CompressionCodecName.ZSTD
     private[scio] val DefaultConfiguration = null
     private[scio] val DefaultShardNameTemplate = null
     private[scio] val DefaultTempDirectory = null

scio-parquet/src/main/scala/com/spotify/scio/parquet/tensorflow/ParquetExampleIO.scala

@@ -221,7 +221,7 @@ object ParquetExampleIO {
   object WriteParam {
     private[tensorflow] val DefaultNumShards = 0
     private[tensorflow] val DefaultSuffix = ".parquet"
-    private[tensorflow] val DefaultCompression = CompressionCodecName.GZIP
+    private[tensorflow] val DefaultCompression = CompressionCodecName.ZSTD
     private[tensorflow] val DefaultConfiguration = null
     private[tensorflow] val DefaultShardNameTemplate = null
     private[tensorflow] val DefaultTempDirectory = null

scio-parquet/src/main/scala/com/spotify/scio/parquet/types/ParquetTypeIO.scala

@@ -188,7 +188,7 @@ object ParquetTypeIO {
   object WriteParam {
     private[scio] val DefaultNumShards = 0
     private[scio] val DefaultSuffix = ".parquet"
-    private[scio] val DefaultCompression = CompressionCodecName.GZIP
+    private[scio] val DefaultCompression = CompressionCodecName.ZSTD
     private[scio] val DefaultConfiguration = null
     private[scio] val DefaultShardNameTemplate = null
     private[scio] val DefaultTempDirectory = null

scio-smb/src/main/java/org/apache/beam/sdk/extensions/smb/ParquetAvroFileOperations.java

@@ -51,7 +51,7 @@
  * Avro records.
  */
 public class ParquetAvroFileOperations<ValueT> extends FileOperations<ValueT> {
-  static final CompressionCodecName DEFAULT_COMPRESSION = CompressionCodecName.GZIP;
+  static final CompressionCodecName DEFAULT_COMPRESSION = CompressionCodecName.ZSTD;
   private final SerializableSchemaSupplier schemaSupplier;
   private final CompressionCodecName compression;
   private final SerializableConfiguration conf;

scio-smb/src/main/scala/org/apache/beam/sdk/extensions/smb/ParquetTypeFileOperations.scala

@@ -33,7 +33,7 @@ import org.apache.parquet.hadoop.metadata.CompressionCodecName
 import java.nio.channels.{ReadableByteChannel, WritableByteChannel}
 
 object ParquetTypeFileOperations {
-  val DefaultCompression = CompressionCodecName.GZIP
+  val DefaultCompression = CompressionCodecName.ZSTD
   val DefaultConfiguration: Configuration = null
 
   def apply[T: Coder: ParquetType](): ParquetTypeFileOperations[T] = apply(DefaultCompression)

scio-smb/src/test/java/org/apache/beam/sdk/extensions/smb/JsonFileOperationsTest.java

@@ -47,7 +47,7 @@ public void testUncompressed() throws Exception {
 
   @Test
   public void testCompression() throws Exception {
-    test(Compression.GZIP);
+    test(Compression.ZSTD);
   }
 
   private void test(Compression compression) throws Exception {

scio-smb/src/test/java/org/apache/beam/sdk/extensions/smb/ParquetAvroFileOperationsTest.java

@@ -217,13 +217,13 @@ public void testDisplayData() {
     MatcherAssert.assertThat(
         displayData, hasDisplayItem("compression", Compression.UNCOMPRESSED.toString()));
     MatcherAssert.assertThat(
-        displayData, hasDisplayItem("compressionCodecName", CompressionCodecName.GZIP.name()));
+        displayData, hasDisplayItem("compressionCodecName", CompressionCodecName.ZSTD.name()));
     MatcherAssert.assertThat(displayData, hasDisplayItem("schema", USER_SCHEMA.getFullName()));
   }
 
   private void writeFile(ResourceId file) throws IOException {
     final ParquetAvroFileOperations<GenericRecord> fileOperations =
-        ParquetAvroFileOperations.of(USER_SCHEMA, CompressionCodecName.GZIP);
+        ParquetAvroFileOperations.of(USER_SCHEMA, CompressionCodecName.ZSTD);
     final FileOperations.Writer<GenericRecord> writer = fileOperations.createWriter(file);
     for (GenericRecord record : USER_RECORDS) {
       writer.write(record);

scio-smb/src/test/java/org/apache/beam/sdk/extensions/smb/TensorFlowFileOperationsTest.java

@@ -53,7 +53,7 @@ public void testUncompressed() throws Exception {
 
   @Test
   public void testCompressed() throws Exception {
-    test(Compression.GZIP);
+    test(Compression.ZSTD);
   }
 
   private void test(Compression compression) throws Exception {

site/src/main/paradox/io/Parquet.md

@@ -287,7 +287,9 @@ val parquetConf: Configuration = {
 Here are some other recommended settings.
 
 - `numShards` - This should be explicitly set so that the size of each output file is smaller than but close to `parquet.block.size`, i.e. 1 GiB. This guarantees that each file contains 1 row group only and reduces seeks.
-- `compression` - `SNAPPY` and `GZIP` work out of the box. Snappy is less CPU intensive but has lower compression ratio. In our benchmarks GZIP seem to work better on GCS.
+- `compression` - Parquet defaults to ZSTD compression with a level of 3; compression level can be set to any integer from 1-22 using the configuration option `parquet.compression.codec.zstd.level`. `SNAPPY` and `GZIP` compression types also work out of the box; Snappy is less CPU intensive but has lower compression ratio. In our benchmarks GZIP seem to work better than Snappy on GCS.
+
+A full list of Parquet configuration options can be found [here](https://github.com/apache/parquet-mr/blob/master/parquet-hadoop/README.md).
 
 ## Parquet Reads in Scio 0.12.0+