A style guide for writers of new reusable PTransforms.
Be consistent with prior art:
So you want to develop a library that people will use in their Beam pipelines - a connector to a third-party system, a machine learning algorithm, etc. How should you expose it?
Do:
PTransform. This allows the structure of the transform to evolve transparently to the code that uses it: e.g. something that started as a ParDo can become a more complex transform over time.CombineFn, or nontrivial DoFns (those that are more than just a single @ProcessElement method). As a rule of thumb: expose these if you anticipate that the full packaged PTransform may be insufficient for a user's needs and the user may want to reuse the lower-level primitive.Do not:
DoFn, concrete Source or Sink classes, etc., in order to avoid presenting users with a confusing choice between applying the PTransform or using the DoFn/Source/Sink.Do:
PascalCase in Java and Python, functions in camelCase in Java but snake_case in Python, etc.MongoDbIO.read(), Flatten.iterables().PTransform classes also like verbs (e.g.: MongoDbIO.Read, Flatten.Iterables).MongoDbIO, JdbcIO.Do not:
transform, source, sink, reader, writer, bound, unbound in PTransform class names (note: bounded and unbounded are fine when referring to whether a PCollection is bounded or unbounded): these words are redundant, confusing, obsolete, or name an existing different concept in the SDK.PCollection: anything of which there may be a very large number of instances (if there can be >1000 of it, it should be in a PCollection), or which is potentially not known at pipeline construction time. E.g.: records to be processed or written to a third-party system; filenames to be read. Exception: sometimes Beam APIs require things to be known at pipeline construction time - e.g. the Bounded/UnboundedSource API. If you absolutely have to use such an API, its input can of course go only into transform configuration.ValueProviders) and does not depend on the contents of the transform's input PCollections. E.g.: a database query or connection string; credentials; a user-specified callback; a tuning parameter. One advantage of putting a parameter into transform configuration is, it can be validated at pipeline construction time.Do:
Do not:
JdbcIO. Exception 1: if some parameters of the underlying library interact with Beam semantics non-trivially, then expose them. E.g. when developing a connector to a pub/sub system that has a “delivery guarantee” parameter for publishers, expose the parameter but prohibit values incompatible with the Beam model (at-most-once and exactly-once). Exception 2: if the underlying library’s configuration class is cumbersome to use - e.g. does not declare a stable API, exposes problematic transitive dependencies, or does not obey semantic versioning - in this case, it is better to wrap it and expose a cleaner and more stable API to users of the transform.Detect errors early. Errors can be detected at the following stages:
For example:
URL class, rather than the String class.PCollection<Document> rather than PCollection<String> (assuming it is possible to provide a Coder for Document).Favor data consistency above everything else. Do not mask data loss or corruption. If data loss can't be prevented, fail.
Do:
DoFn, retry transient failures if the operation is likely to succeed on retry. Perform such retries at the narrowest scope possible in order to minimize the amount of retried work (i.e. ideally at the level of the RPC library itself, or at the level of directly sending the failing RPC to a third-party system). Otherwise, let the runner retry work at the appropriate level of granularity for you (different runners may have different retry behavior, but most of them do some retrying).Do not:
catch(...) { log an error; return null or false or otherwise ignore; } Rule of thumb: if a bundle didn't fail, its output must be correct and complete. For a user, a transform that logged an error but succeeded is silent data loss.Many runners optimize chains of ParDos in ways that improve performance if the ParDos emit a small to moderate number of elements per input element, or have relatively cheap per-element processing (e.g. Dataflow's “fusion”, Apex “compute locality”), but limit parallelization if these assumptions are violated. In that case you may need a “fusion break” (Reshuffle.of()) to improve the parallelizability of processing the output PCollection of the ParDo.
ParDo that outputs a potentially large number of elements per input element, apply a fusion break after this ParDo to make sure downstream transforms can process its output in parallel.ParDo that takes a very long time to process an element, insert a fusion break before this ParDo to make sure all or most elements can be processed in parallel regardless of how its input PCollection was produced.Document how to configure the transform (give code examples), and what guarantees it expects about its input or provides about its output, accounting for the Beam model. E.g.:
PCollection with information from an external key-value store), what are the guarantees on freshness of queried data: e.g. is it all loaded at the beginning of the transform, or queried per-element (in that case, what if data for a single element changes while the transform runs)?PCollection and emitting output into the output PCollection, what is this relationship? (e.g. if the transform internally does windowing, triggering, grouping, or uses the state or timers API)Anticipate abnormal situations that a user of the transform may run into. Log information that they would have found sufficient for debugging, but limit the volume of logging. Here is some advice that applies to all programs, but is especially important when data volume is massive and execution is distributed.
Do:
Do not:
INFO per element or per bundle. DEBUG/TRACE may be okay because these levels are disabled by default.Data processing is tricky, full of corner cases, and difficult to debug, because pipelines take a long time to run, it‘s hard to check if the output is correct, you can’t attach a debugger, and you often can't log as much as you wish to, due to high volume of data. Because of that, testing is particularly important.
TestPipeline and PAssert. Start with testing against the direct runner. Assertions on PCollection contents should be strict: e.g. when a read from a database is expected to read the numbers 1 through 10, assert not just that there are 10 elements in the output PCollection, or that each element is in the range [1, 10] - but assert that each number 1 through 10 appears exactly once.TestPipeline, extract this logic into unit-testable functions, and unit-test them. Common corner cases are:DoFns processing empty bundlesDoFns processing extremely large bundles (contents doesn't fit in memory, including “hot keys” with a very large number of values)Closeable/AutoCloseable resources in failure casesBoundedSource.split (e.g. splitting key or offset ranges), iteration over empty data sources or composite data sources that have some empty components.DoFns, CombineFns, and BoundedSources, consider using DoFnTester, CombineFnTester, and SourceTestUtils respectively which can exercise the code in non-trivial ways to flesh out potential bugs.TestStream.The code for constructing and validating a transform is usually trivial and mostly boilerplate. However, minor mistakes or typos in it can have serious consequences (e.g. ignoring a property that the user has set), so it needs to be tested as well. Yet, an excessive amount of trivial tests can be hard to maintain and give a false impression that the transform is well-tested.
Do:
withFoo() and withBar() cannot both be specified at the same time, test that a transform specifying both of them is rejected, rather than one of the properties being silently ignored at runtime.withFoo() method (including each overload) has effect (is not ignored), using TestPipeline and PAssert to create tests where the expected test results depend on the value of withFoo().Do not:
Do:
@Experimental (Java) or @experimental ([Python]({{ site.baseurl }}/documentation/sdks/pydoc/{{ site.release_latest }}/apache_beam.utils.annotations.html)).@Deprecated (Java) or @deprecated ([Python]({{ site.baseurl }}/documentation/sdks/pydoc/{{ site.release_latest }}/apache_beam.utils.annotations.html)).Do not:
Good examples for most of the practices below are JdbcIO and MongoDbIO.
Whenever possible, use types specific to the nature of the transform. People can wrap it with conversion DoFns from their own types if necessary. E.g. a Datastore connector should use the Datastore Entity type, a MongoDb connector should use Mongo Document type, not a String representation of the JSON.
Sometimes that's not possible (e.g. JDBC does not provide a Beam-compatible (encodable with a Coder) “JDBC record” datatype) - then let the user provide a function for converting between the transform-specific type and a Beam-compatible type (e.g. see JdbcIO and MongoDbGridFSIO).
When the transform should logically return a composite type for which no Java class exists yet, create a new POJO class with well-named fields. Do not use generic tuple classes or KV (unless the fields are legitimately a key and a value).
If the transform needs to return multiple collections, it should be a PTransform<..., PCollectionTuple> and expose methods getBlahTag() for each collection.
E.g. if you want to return a PCollection<Foo> and a PCollection<Bar>, expose TupleTag<Foo> getFooTag() and TupleTag<Bar> getBarTag().
For example:
public class MyTransform extends PTransform<..., PCollectionTuple> { private final TupleTag<Moo> mooTag = new TupleTag<Moo>() {}; private final TupleTag<Blah> blahTag = new TupleTag<Blah>() {}; ... PCollectionTuple apply(... input) { ... PCollection<Moo> moo = ...; PCollection<Blah> blah = ...; return PCollectionTuple.of(mooTag, moo) .and(blahTag, blah); } public TupleTag<Moo> getMooTag() { return mooTag; } public TupleTag<Blah> getBlahTag() { return blahTag; } ... }
Make the transform class immutable, with methods to produce modified immutable objects. Use AutoValue. Autovalue can provide a Builder helper class. Use @Nullable to mark parameters of class type that don't have a default value or whose default value is null, except for primitive types (e.g. int).
@AutoValue public abstract static class MyTransform extends PTransform<...> { int getMoo(); @Nullable abstract String getBlah(); abstract Builder toBuilder(); @AutoValue.Builder abstract static class Builder { abstract Builder setMoo(int moo); abstract Builder setBlah(String blah); abstract MyTransform build(); } ... }
Provide a single argumentless static factory method, either in the enclosing class (see “Packaging a family of transforms”) or in the transform class itself.
public class Thumbs { public static Twiddle twiddle() { return new AutoValue_Thumbs_Twiddle.Builder().build(); } public abstract static class Twiddle extends PTransform<...> { ... } } // or: public abstract static class TwiddleThumbs extends PTransform<...> { public static TwiddleThumbs create() { return new AutoValue_Thumbs_Twiddle.Builder().build(); } ... }
Exception: when transform has a single overwhelmingly most important parameter, then call the factory method of and put the parameter into an argument of the factory method: ParDo.of(DoFn).withAllowedLateness().
Call them withBlah(). All builder methods must return exactly the same type; if it's a parameterized (generic) type, with the same values of type parameters.
Treat withBlah() methods as an unordered set of keyword arguments - result must not depend on the order in which you call withFoo() and withBar() (e.g., withBar() must not read the current value of foo).
Document implications of each withBlah method: when to use this method at all, what values are allowed, what is the default, what are the implications of changing the value.
/** * Returns a new {@link TwiddleThumbs} transform with moo set * to the given value. * * <p>Valid values are 0 (inclusive) to 100 (exclusive). The default is 42. * * <p>Higher values generally improve throughput, but increase chance * of spontaneous combustion. */ public Twiddle withMoo(int moo) { checkArgument(moo >= 0 && moo < 100, "Thumbs.Twiddle.withMoo() called with an invalid moo of %s. " + "Valid values are 0 (inclusive) to 100 (exclusive)", moo); return toBuilder().setMoo(moo).build(); }
Specify them in the factory method (factory method returns an object with default values).
public class Thumbs { public static Twiddle twiddle() { return new AutoValue_Thumbs_Twiddle.Builder().setMoo(42).build(); } ... }
If several parameters of the transform are very tightly logically coupled, sometimes it makes sense to encapsulate them into a container object. Use the same guidelines for this container object (make it immutable, use AutoValue with builders, document withBlah() methods, etc.). For an example, see JdbcIO.DataSourceConfiguration.
All type parameters should be specified explicitly on factory method. Builder methods (withBlah()) should not change the types.
public class Thumbs { public static Twiddle<T> twiddle() { return new AutoValue_Thumbs_Twiddle.Builder<T>().build(); } @AutoValue public abstract static class Twiddle<T> extends PTransform<PCollection<Foo>, PCollection<Bar<T>>> { … @Nullable abstract Bar<T> getBar(); abstract Builder<T> toBuilder(); @AutoValue.Builder abstract static class Builder<T> { … abstract Builder<T> setBar(Bar<T> bar); abstract Twiddle<T> build(); } … } } // User code: Thumbs.Twiddle<String> twiddle = Thumbs.<String>twiddle(); // Or: PCollection<Bar<String>> bars = foos.apply(Thumbs.<String>twiddle() … );
Exception: when the transform has a single most important parameter and this parameter depends on type T, then prefer to put it right into the factory method: e.g. Combine.globally(SerializableFunction<Iterable<V>,V>). This improves Java's type inference and allows the user not to specify type parameters explicitly.
When the transform has more than one type parameter, or if the meaning of the parameter is non-obvious, name the type parameters like SomethingT, e.g.: a PTransform implementing a classifier algorithm and assigning each input element with a label might be typed as Classify<InputT, LabelT>.
If the transform has an aspect of behavior to be customized by a user's code, make a decision as follows:
Do:
PTransform, then the transform itself should not be extensible. E.g., a transform that writes JSON objects to a third-party system should take a PCollection<JsonObject> (assuming it is possible to provide a Coder for JsonObject), rather than taking a generic PCollection<T> and a SerializableFunction<T, JsonObject> (anti-example that should be fixed: TextIO).SerializableFunction or define your own serializable function-like type (ideally single-method, for interoperability with Java 8 lambdas). Because Java erases the types of lambdas, you should be sure to have adequate type information even if a raw-type SerializableFunction is provided by the user. See MapElements and FlatMapElements for examples of how to use SimpleFunction and SerializableFunction in tandem to support Java 7 and Java 8 well.Do not:
PTransform class.When developing a family of highly related transforms (e.g. interacting with the same system in different ways, or providing different implementations of the same high-level task), use a top-level class as a namespace, with multiple factory methods returning transforms corresponding to each individual use case.
The container class must have a private constructor, so it can't be instantiated directly.
Document common stuff at BlahIO level, and each factory method individually.
/** Transforms for clustering data. */ public class Cluster { // Force use of static factory methods. private Cluster() {} /** Returns a new {@link UsingKMeans} transform. */ public static UsingKMeans usingKMeans() { ... } public static Hierarchically hierarchically() { ... } /** Clusters data using the K-Means algorithm. */ public static class UsingKMeans extends PTransform<...> { ... } public static class Hierarchically extends PTransform<...> { ... } } public lass FooIO { // Force use of static factory methods. private FooIO() {} public static Read read() { ... } ... public static class Read extends PTransform<...> { ... } public static class Write extends PTransform<...> { ... } public static class Delete extends PTransform<...> { ... } public static class Mutate extends PTransform<...> { ... } }
When supporting multiple versions with incompatible APIs, use the version as a namespace-like class too, and put implementations of different API versions in different files.
// FooIO.java public class FooIO { // Force use of static factory methods. private FooIO() {} public static FooV1 v1() { return new FooV1(); } public static FooV2 v2() { return new FooV2(); } } // FooV1.java public class FooV1 { // Force use of static factory methods outside the package. FooV1() {} public static Read read() { ... } public static class Read extends PTransform<...> { ... } } // FooV2.java public static class FooV2 { // Force use of static factory methods outside the package. FooV2() {} public static Read read() { ... } public static class Read extends PTransform<...> { ... } }
ImmutableList).PCollections must be immutable.DoFn, PTransform, CombineFn and other instances will be serialized. Keep the amount of serialized data to a minimum: Mark fields that you don‘t want serialized as transient. Make classes static whenever possible (so that the instance doesn’t capture and serialize the enclosing class instance). Note: In some cases this means that you cannot use anonymous classes.
.withBlah() methods using checkArgument(). Error messages should mention the name of the parameter, the actual value, and the range of valid values.PTransform's .expand() method.PTransform takes from PipelineOptions in the PTransform‘s .validate(PipelineOptions) method. These validations will be executed when the pipeline is already fully constructed/expanded and is about to be run with a particular PipelineOptions. Most PTransforms do not use PipelineOptions and thus don’t need a validate() method - instead, they should perform their validation via the two other methods above.@AutoValue public abstract class TwiddleThumbs extends PTransform<PCollection<Foo>, PCollection<Bar>> { abstract int getMoo(); abstract String getBoo(); ... // Validating individual parameters public TwiddleThumbs withMoo(int moo) { checkArgument( moo >= 0 && moo < 100, "Moo must be between 0 (inclusive) and 100 (exclusive), but was: %s", moo); return toBuilder().setMoo(moo).build(); } public TwiddleThumbs withBoo(String boo) { checkArgument(boo != null, "Boo can not be null"); checkArgument(!boo.isEmpty(), "Boo can not be empty"); return toBuilder().setBoo(boo).build(); } @Override public void validate(PipelineOptions options) { int woo = options.as(TwiddleThumbsOptions.class).getWoo(); checkArgument( woo > getMoo(), "Woo (%s) must be smaller than moo (%s)", woo, getMoo()); } @Override public PCollection<Bar> expand(PCollection<Foo> input) { // Validating that a required parameter is present checkArgument(getBoo() != null, "Must specify boo"); // Validating a combination of parameters checkArgument( getMoo() == 0 || getBoo() == null, "Must specify at most one of moo or boo, but was: moo = %s, boo = %s", getMoo(), getBoo()); ... } }
Coders are a way for a Beam runner to materialize intermediate data or transmit it between workers when necessary. Coder should not be used as a general-purpose API for parsing or writing binary formats because the particular binary encoding of a Coder is intended to be its private implementation detail.
Provide default Coders for all new data types. Use @DefaultCoder annotations or CoderProviderRegistrar classes annotated with @AutoService: see usages of these classes in the SDK for examples. If performance is not important, you can use SerializableCoder or AvroCoder. Otherwise, develop an efficient custom coder (subclass AtomicCoder for concrete types, StructuredCoder for generic types).
All PCollections created by your PTransform (both output and intermediate collections) must have a Coder set on them: a user should never need to call .setCoder() to “fix up” a coder on a PCollection produced by your PTransform (in fact, Beam intends to eventually deprecate setCoder). In some cases, coder inference will be sufficient to achieve this; in other cases, your transform will need to explicitly call setCoder on its collections.
If the collection is of a concrete type, that type usually has a corresponding coder. Use a specific most efficient coder (e.g. StringUtf8Coder.of() for strings, ByteArrayCoder.of() for byte arrays, etc.), rather than a general-purpose coder like SerializableCoder.
If the type of the collection involves generic type variables, the situation is more complex:
PCollection, available via input.getCoder().input.getPipeline().getCoderRegistry().getCoder(TypeDescriptor). Use utilities in TypeDescriptors to obtain the TypeDescriptor for the generic type. For an example of this approach, see the implementation of AvroIO.parseGenericRecords(). However, coder inference for generic types is best-effort and in some cases it may fail due to Java type erasure.Coder for the relevant type variable(s) as a configuration parameter of your PTransform. (e.g. AvroIO.<T>parseGenericRecords().withCoder(Coder<T>)). Fall back to inference if the coder was not explicitly specified.