Getting started with Apache Beam

Info transforms

Some transforms like Stream Lookup read data from other transforms. This is handled by side-inputs for the data in the Beam API and is as such fully supported.

Target transforms

Sometimes you want to target specific transforms like in Switch Case or Filter Rows. This is fully supported as well and handled by the Beam API which handles additional outputs.

Non-Beam input transforms

When you’re reading data using a non-beam transform (see Beam specific transforms above) we need to make sure that this transform is executed in exactly one thread. Otherwise, you might read your XML or JSON document many times by the inherently parallel intentions of the various engines. This is handled by doing a Group By over a single value. You’ll see the following in for example your Dataflow pipeline: Create.Values → WithKeys → GroupByKey → Values → Flatten.Iterables → ParDo …​ and all this is just done to make sure we only ever execute our transform once.

Non-Beam Output transforms

The insistence of a Beam pipeline to run work in an embarrassingly parallel way can trip you up on the output side. In general, it is NOT possible to limit the number of copies of a particular transform. What Hop tries to do is perform a series of operations to try and force a single thread. However, this does not work on all runners. Flink for example insists on doing even this in parallel: GroupByKey(Void) → Values() → Flatten().

You can set the number of copies string of a transform to include SINGLE_BEAM (click on the transform and select Number of copies in the Hop GUI).

A better way to deal with this problem is NOT to insist on creating a single file. You can include the unique ID of a transform in the filename with variable ${Internal.Transform.ID}. This will allow the various transform copies to write unique files to the same folder.

Row batching with non-Beam transforms

A lot of target databases like to receive rows in batches of records. So if you have a transform like for example Table Output or Neo4j Output you might see that performance is not that great. This is because by default the Beam programming model is designed to stream rows of data through a pipeline in bundles and the Hop API only knows about a single record at once. For these transforms you can include BATCH in the number of copies string of a transform click on the transform and select Number of copies in the Hop GUI). For these flagged transforms you can then specify 2 parameters in your Beam pipeline run configurations. When you set these you can determine how long rows are kept behind before being forced to the transforms in question

Streaming Hop transform flush interval: how long in time are rows kept and batched up? If you care about latency make this lower (500 or lower). If you have a long-running batching pipeline, make it higher (10000 or higher perhaps).

Hop streaming transforms buffer size: how many rows are being batched? Consider making it the same as the batching size you use in your transform metadata (e.g. Table Output, Neo4j Cypher, …​)

Please note that these are maximum values. If the end of a bundle is reached in a pipeline rows are always forced to the transform code and as such pushed to the target system. To get an idea of how many times a batching buffer is flushed to the underlying transform code (and as such to for example a remote database) we added a Flushes metric. You will notice this in your metrics view in the Hop GUI when executing.