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 parallel can also trip you up on the output side. In rare cases maybe you don’t want a server to be bombarded by dozens of inbound connections. To limit the amount of output copies you can include SINGLE_BEAM in the number of copies value of a transform (click on the transform and select Number of copies in the Hop GUI). This will do a GroupBy exercise over all records to iterate over those and force a single thread.

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.