[BEAM-701] Support Avro write compression #1038

ravwojdyla · 2016-10-03T18:44:17Z

Be sure to do all of the following to help us incorporate your contribution
quickly and easily:

Make sure the PR title is formatted like:
[BEAM-<Jira issue #>] Description of pull request
Make sure tests pass via mvn clean verify. (Even better, enable
Travis-CI on your fork and ensure the whole test matrix passes).
Replace <Jira issue #> in the title with the actual Jira issue
number, if there is one.
If this contribution is large, please file an Apache
Individual Contributor License Agreement.

dhalperi

Left some design feedback. Thoughts?

dhalperi · 2016-10-03T19:43:41Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

@@ -443,13 +445,24 @@ private Read() {}
    }

    /**
+     * Returns a {@link PTransform} that writes Avro file(s) using specified codec.
+     * Codec cna be a string value drawn from those in


Drop L449 and 450? This function seems like it does not take a string.

(typo: can)

dhalperi · 2016-10-03T19:46:36Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/util/package-info.java

+ */
+
+/** Defines utilities that ease working with IO in Apache Beam. */
+package org.apache.beam.sdk.io.util;


Generally trying to avoid util packages. Any reason not to inline this code into AvroIO?

My reasoning: AvroIO is already pretty big and complex class, adding another class would complicate it even further, making it even less human readable.

@dhalperi therefor, is it fine if we keep this as a separate class, it could be in a different package than util if needed?

can we move SerializableAvroCodecFactory inside of the io package and make it package-private? That would probably be easiest.

I've found that anything that appears to be part of the public API ends up having users that depend on it, even if we mark it as for internal use only. So the best solution is to use Java visibility to make sure it's a private utility.

dhalperi · 2016-10-03T21:58:57Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/util/SerializableAvroCodecFactory.java

+ * A wrapper allows  {@link org.apache.avro.file.CodecFactory}s to be serialized using Java's
+ * standard serialization mechanisms.
+ */
+public class SerializableAvroCodecFactory implements Externalizable {


I am generally saddened that Avro's CodecFactory is so unusable.

Suppose I proposed we made AvroIO simply accept a String codecName and that the code simply uses CodecFactory.fromString(codecName) at execution time? I guess you'd argue that this would break xz and deflate outside of the default levels.

Is that important?

If so, can we move the list of valid strings / etc. inside of AvroIO?

I see your point. I would argue using String for configuration is error prone. Furthermore while setting or retrieving codec, from user perspective she can only see CodecFactory - it's only inside AvroIO that we use SerializableAvroCodecFactory - so the whole thing is hidden from the user. Also it's not that big of an overhead - both logic and code wise.

What do you mean by If so, can we move the list of valid strings / etc. inside of AvroIO??

dhalperi · 2016-10-03T22:00:12Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/util/SerializableAvroCodecFactory.java

+
+  public SerializableAvroCodecFactory(CodecFactory codecFactory) {
+    checkNotNull(codecFactory, "Codec can't be null");
+    checkState(checkIsSupportedCodec(codecFactory), "%s is not supported", codecFactory.toString());


no .toString() on stuff using checkState -- it will do that automatically in the case that it prints the message, but will also save cycles by not toStringing in the success case.

ravwojdyla · 2016-10-04T14:54:42Z

@dhalperi thanks for review. Added comments + some fixes.

dhalperi

LGTM modulo moving the utility into the IO package for proper visibility.

dhalperi · 2016-10-07T21:08:03Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

     * A {@link PTransform} that writes a bounded {@link PCollection} to an Avro file (or
     * multiple Avro files matching a sharding pattern).
     *
     * @param <T> the type of each of the elements of the input PCollection
     */
    public static class Bound<T> extends PTransform<PCollection<T>, PDone> {
      private static final String DEFAULT_SHARD_TEMPLATE = ShardNameTemplate.INDEX_OF_MAX;
+      private static final SerializableAvroCodecFactory DEFAULT_CODEC =
+          new SerializableAvroCodecFactory(CodecFactory.deflateCodec(6));


Is this the current default, or a behavior change? (I assume the actual current default is that we use the null codec?) If so, put this behavior change in the commit message :)

dhalperi · 2016-10-07T21:11:28Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/util/package-info.java

+ */
+
+/** Defines utilities that ease working with IO in Apache Beam. */
+package org.apache.beam.sdk.io.util;


can we move SerializableAvroCodecFactory inside of the io package and make it package-private? That would probably be easiest.

I've found that anything that appears to be part of the public API ends up having users that depend on it, even if we mark it as for internal use only. So the best solution is to use Java visibility to make sure it's a private utility.

dhalperi · 2016-10-07T21:13:07Z

sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroIOTest.java

+    AvroIO.Write.Bound<GenericRecord> serdeWrite = SerializableUtils.clone(write);
+
+    assertTrue(serdeWrite.getCodec().toString().equals(CodecFactory.xzCodec(9).toString()));
+  }


This is _completely optional. I'm only including this nitpick because there were other actual comments_

We'll get slightly better error message if the code was written as:

assertEquals(CodecFactory.xzCodec(9), serdWrite.getCodec())`

(with toString() as it is now, if those equality checks don't work).

There are a few other optional instances of opportunities for improvement in this file and the other test file.

Fixed. What are the other improvements?

ping @dhalperi

BEHAVIOUR CHANGE: prior to this change Avro output would not use compression. Starting from this commit, by default Avro output is compressed using deflate codec (level 6).

ravwojdyla mentioned this pull request Oct 3, 2016

Avro compression GoogleCloudPlatform/DataflowJavaSDK#461

Closed

ravwojdyla force-pushed the avro_compression branch from 68ceef1 to 172c112 Compare October 3, 2016 20:08

dhalperi requested changes Oct 3, 2016

View reviewed changes

ravwojdyla force-pushed the avro_compression branch from 172c112 to a167bca Compare October 4, 2016 14:42

dhalperi requested changes Oct 7, 2016

View reviewed changes

ravwojdyla force-pushed the avro_compression branch 2 times, most recently from eab8d8e to 7d47d37 Compare October 8, 2016 00:27

Add compression codec for AvroIO

e5e91d3

BEHAVIOUR CHANGE: prior to this change Avro output would not use compression. Starting from this commit, by default Avro output is compressed using deflate codec (level 6).

ravwojdyla force-pushed the avro_compression branch 5 times, most recently from 1a6a1db to 9c954c7 Compare October 8, 2016 01:50

Add tests for Avro compression codec

449561c

ravwojdyla force-pushed the avro_compression branch from 9c954c7 to 449561c Compare October 8, 2016 01:50

nevillelyh mentioned this pull request Oct 12, 2016

Write compressed files GoogleCloudPlatform/DataflowJavaSDK#404

Closed

asfgit closed this in 142229e Oct 12, 2016

ravwojdyla deleted the avro_compression branch October 13, 2016 22:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-701] Support Avro write compression #1038

[BEAM-701] Support Avro write compression #1038

ravwojdyla commented Oct 3, 2016 •

edited

dhalperi left a comment

dhalperi Oct 3, 2016

dhalperi Oct 3, 2016

ravwojdyla Oct 4, 2016

dhalperi Oct 3, 2016

ravwojdyla Oct 4, 2016

ravwojdyla Oct 7, 2016

dhalperi Oct 7, 2016

ravwojdyla Oct 8, 2016

dhalperi Oct 3, 2016 •

edited

ravwojdyla Oct 4, 2016

dhalperi Oct 3, 2016

ravwojdyla Oct 4, 2016

ravwojdyla commented Oct 4, 2016

dhalperi left a comment

dhalperi Oct 7, 2016

ravwojdyla Oct 8, 2016

dhalperi Oct 7, 2016

dhalperi Oct 7, 2016 •

edited

ravwojdyla Oct 8, 2016

ravwojdyla Oct 12, 2016

[BEAM-701] Support Avro write compression #1038

[BEAM-701] Support Avro write compression #1038

Conversation

ravwojdyla commented Oct 3, 2016 • edited

dhalperi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dhalperi Oct 3, 2016 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ravwojdyla commented Oct 4, 2016

dhalperi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dhalperi Oct 7, 2016 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ravwojdyla commented Oct 3, 2016 •

edited

dhalperi Oct 3, 2016 •

edited

dhalperi Oct 7, 2016 •

edited