Watermark architecture proposal for Spark Structured Streaming framework

🕥 7 min.

The multiple aggregations problem with Spark Structured Streaming framework

Developing the translation layer (called runner) from Apache Beam to Apache Spark we faced an issue with the Spark Structured Streaming framework: the problem is that this framework does not support more than one aggregation in a streaming pipeline. For example, you cannot do a group by then a reduce by in a streaming pipeline. There is an open ticket in the Spark project, an ongoing design and an ongoing PR, but, as for now, they received no update since the summer 2019. As a consequence, the Beam runner based on this framework is on hold waiting for this feature from the Spark project.

The underlying problem: the global watermark in the Spark Structured Streaming framework

Before explaining this problem, let's explain what a watermark is

What is a watermark?

In streaming systems, there is always lags between the time at which a given element was produced (the event timestamp) and the time at which it was received by the system (the processing timestamp) because there can be network outages or slow downs etc... There also can be out of order data. 

To deal with these two facts, streaming systems define the notion of watermark. It is what gives the system the notion of completeness of data in a constant flow of streaming data. It is the point in time when the system should not receive older elements. As streaming systems rely on windowing to divide this stream of data, the watermark can also be defined as the system notion of when all the data in a certain window can be expected to have arrived in the streaming pipeline. When the watermark passes the end of the window, the system outputs data.

If we take an example of a very simple static naïve watermark defined as an offset of 10 min: 

    at 2:10 the system considers that no element with a timestamp older than 2:00 can arrive

A corollary notion is late data. It is data that arrives behind the watermark. For example, with the previous watermark example, let's say that an element with timestamp of 1:59 arrives at 2:11. At 2:11 the watermark tells the system that no element with timestamp older that 2:01 (offset of 10 min) can arrive, so this element is behind the watermark, so it will be considered late by the system. As the consequence, the system should drop it. Still, we can define a tolerance that we call allowed lateness that could permit to process this late element. For this element not to be dropped, we would need to set an allowed lateness value of at least 2 min (the element arrives 12 min after it was produced - watermark of 10 min offset).

The problem with the global watermark in an example

Back to the problem: Spark Structured Streaming framework does not support more than one aggregation in a streaming pipeline because the watermark scope in Spark Structured Streaming framework is global for the whole pipeline, there is no watermark per operation. At the end of each microbatch the global watermark is updated with the maximum timestamp received. How is it a problem? Let's answer in an example:

Consider we have a streaming pipeline with 2 aggregations (Op1 and Op2 in blue in the diagram below). The first aggregation outputs the highest valued element out of a window of 3 seconds.

Now let's say that the source sends 3 elements with value 6, 4 and 5 and timestamps 1, 2 and 3. These elements get buffered inside Op1 and Op1 updates the global watermark to value 3 (the maximum timestamp seen). As the watermark reached the end of Op1's window (window of 3 seconds), then it is time for Op1 to output its data. The highest valued element is value 6 but timestamp 1. As a consequence this element will be considered by Op2 as late because Op2 relies on the global watermark which value is 3, so the element's timestamp of 1 is behind the global watermark. As a consequence, this element will be dropped by Op2 leading to incorrect results.

So, to avoid that incorrect results could be produced by downstream operations, Spark Structured Streaming framework deactivates the support for more than one aggregation in streaming pipelines.

A possible solution to this problem

A watermark per operation

A possible solution to this problem is to replace the global watermark by a watermark per transform.

And the watermarks values will be propagated from the source through the different operations of the pipeline. In fact, in most streaming systems, the watermark is a special type of element that flows with the data downstream the pipeline. The local watermark values will be updated as new watermarks and data arrive. For each operation we need to define:

• an InputWatermark = Min {outputWatermark(previous operations)} *

• an OutputWatermark = Min {inputWatermark, oldest(processed element)}

* In a straight pipeline, the InputWatermark is simply the OutputWatermark of the previous operation

Updated example

Let's update the previous example of the streaming pipeline with Op1 and Op2. Only now, there is no more global watermark, we define a watermark per operation in green in the diagram below:

As said above, the watermark is set by the source. Let's say that the source tells the system that no element older than timestamp 1 can arrive (source watermark = 1). This value is propagated to the input watermark and to the ouput watermark of Op1 according to the rules defined in the previous paragraph. The watermarks of Op2 are updated as well according to the same rules. We end up having the green values in the above diagram.

Now let's say that we receive the very same 3 elements as before with values 6, 4 and 5 and timestamps 1, 2 and 3. 

They get buffered in Op1. Op1 updates its oldest processed counter (in red) to the minimum timestamp seen. And, as the output watermark is the minimum between this counter and the input watermark, it remains set to 1. And, as the ouput watermark of Op1 is not updated, the input and output watermarks of Op2 are not either.

Let's say now that the source updates its watermark to 3 (in red in the diagram below), saying that it should not receive older elements than timestamp 3.

So the input watermark of Op1 is updated to the output watermark of the previous step (the source) but the output watermark of Op1 is not updated as it is the minimum between the oldest processed counter (which is 1) and the input watermark. As the output watermark of Op1 is not updated, the input and output watermarks of Op2 are not either.

But something happens now in Op1: its input watermark is now value 3, so it has passed the end of the window. Therefore it is time for Op1 to output its data and clear its buffers. So it outputs the highest element of value 6 and timestamp 1 as shown below.

But now, this element is no more dropped by Op2 because the input watermark of Op2 is still 1, Op2 no more considers the element late. It just processes it, buffers it and updates its oldest processed counter.

Now, one final thing you may wonder, is when and how does Op2 updates its watermarks?

Well, after outputting its data,  Op1 updates its output watermark to 3 and re-initializes its oldest processed counter. After that it sends its output watermark downstream. So Op2 receives the output watermark of Op1 after the element and, as a consequence, updates its input watermark as shown below. This allows to update the watermarks on a timely manner and not wait for the arrival of new data in the pipeline to update watermarks.

There is something left to be discussed: the late case that we will see in the next paragraph.

What happens for late data in this architecture?

A given element is late for a given operation when its timestamp is inferior to the InputWatermark of the operation. As the OutputWatermark of this operation is defined as Min {inputWatermark, oldest(processed element)}, then the OutputWatermark of the operation is set to the timestamp of the element. So, for the next operation, the InputWatermark will be set to the OuputWatermark of the previous operation, and thus will be set to the timestamp of the element. So we can see that the watermarks of all downstream operations will be set to the value of this element's timestamp. So, to sum it up: a late element will delay downstream operations.

Conclusion

This watermark architecture was proposed to the Apache Spark project.

Export metrics from Apache Beam pipelines

🕥 12 min. 

 This blog post is about part of this talk that I gave at the ApacheCon 2018 about universal metrics in Apache Beam. More precisely, it deals with metrics export: how Beam metrics are exported to the outside world in a running Beam pipeline.

What are Beam metrics?

The content below is also available in the Beam website but is recalled here for the completeness of the article

In the Beam model, metrics provide some insight into the current state of a user pipeline, potentially while the pipeline is running. For example, this allows the user to:

• Check the number of errors encountered while running a specific step in the pipeline
• Monitor the number of calls made to an external service
• Count the number of elements that have been processed
• …and so on.

Metrics are: 

• Named: Each metric has a name which consists of a namespace and an actual name. The namespace can be used to differentiate between multiple metrics with the same name and also allows querying for all metrics within a specific namespace.
• Scoped: Each metric is reported against a specific step in the pipeline (i.e. a specific transform in the pipeline), indicating what code was running when the metric was declared. This allows reporting the same metric name in multiple places and identifying the value each transform reported, as well as aggregating the metric across the entire pipeline.
• Dynamically Created: Metrics may be created during runtime without pre-declaring them, in much the same way a logger could be created. This makes it easier to produce metrics in utility code and have them usefully reported.
• Degrade Gracefully: If a runner doesn’t support some part of reporting metrics, the fallback behavior is to drop the metric updates rather than failing the pipeline. If a runner doesn’t support some part of querying metrics, the runner will not return the associated data.
• Attempted/committed metrics: attempted metrics include retrials whereas committed metrics do not: if a bundle (part of Beam PCollection) is retried a counter committed value will not be incremented. If you want some details on retrial and metrics please take a look at slide 18/19 of the related talk. 

Note: It is runner-dependent whether metrics are accessible during pipeline execution or only after jobs have completed.

 

Types of metrics

There are three types of metrics that are supported for the moment: Counter, Distribution and Gauge.

Counter

Counter is a metric that reports a single long value and can be incremented or decremented.

Distribution

Distribution is a metric that reports information about the distribution of reported values (min value, max value, mean value, sum of values and count of values).

Gauge

Gauge is a metric that reports the latest value out of reported values. Since metrics are collected from many workers the value may not be the absolute last, but one of the latest values.

How does a pipeline author use them in Java?

How to query a metric?

Users query metrics using PipelineResult which is Beam object to interact with a pipeline. PipelineResult has a method metrics() which returns a MetricResults object that allows accessing metrics. The main method available in MetricResults allows querying for all metrics matching a given filter.

Complete example

Below, there is a simple example of how to use a Counter metric in a user pipeline.

Metrics export: MetricsPusher

Here comes the core of the subject: the MetricsPusher feature in Beam !

Motivation

Metrics Pusher came to life in the Beam project after these observations:

• Not all execution engines (Spark, Flink etc...)  ship a way to push the metrics to external sinks, they usually rely on their own monitoring UI. Even though the execution engine UI is updated with the Beam metrics, there are some use cases that require to export the metrics values: What if you need to have the metrics in your own application UI ? Or if you need to have the metrics in a metrics backend for reporting ?
• Need for consistency:
• There is no common set of monitoring backend support among execution engines
• There is a difference of availability moment when the pipeline runs: some runners make the metrics available through PipelineResult during the pipeline execution and others only at the end of the pipeline execution
• Beam needs to have a common metrics flow no matter the runners for pipelines to be portable

Architecture

Design principles

Metrics pusher was designed based on these principles:

• No client polling (e.g. JMX) because:
• infrastructure changes (cluster managers, ...) and must not be known of the users
• such pulled metrics would be non-aggregated metrics, that users would need to aggregate
• such pull mechanism raises timing issues: for example if a small batch finishes before the JMX client has had time to pull the metrics, there will be no metric to show.
• Metrics are pushed from the runner and not pulled from the sdk because:
• runners needs to decide when to push in particular because to support Beam committed metrics it needs to push only when a Beam bundle is committed.
• runner system metrics are also defined in Beam (but not yet supported in MetricsPusher) and there again, only the runner knows its internal system.
• Push aggregated (across parallel workers) metrics periodically rather than push event based metrics because:
• aggregated metrics avoid the need for metrics backend to merge metrics.
• runners know how to merge metrics across their workers.
• We chose that Beam manages the metrics sinks IOs mainly for coherence because, as said above, execution engines have different support of the metrics backend.

MetricsPusher architecture

The MetricsPusher service lives as a thread inside the runner that regularly requests for aggregated metrics and then pushes them through a MetricsSink IO to a metrics backend.  

The architecture diagram above focuses on metrics export, if you want more details on the internal runner metric system part, please take a look at part 4 of the related talk; for the purpose of this article let's just say that it is provided as part of the runner-core Beam library so that different runners (Spark, Flink, Dataflow, ...) share the same core metrics architecture.

How to use Metrics pusher ?

There is nothing to code in the user pipeline to use the MetricsPusher feature. The only thing to do is to configure a MetricsSink. If a metrics sink is set up in the configuration, the runner will push metrics to it at a default 5s period.

The configuration is held in the MetricsOptions class. It contains push period configuration and also sink specific options such as type and URL.

As for now, there is only Flink and Spark runners that are supported and the metrics sinks available are:

• MetricsHttpSink: a REST client that sends json serialized metrics to a REST endpoint using an HTTP POST request. Below is an example of the body of the request sending attempted metrics values of a pipeline that defines a metric of each type: 
  
  
• MetricsGraphiteSink: a sink that pushes metrics to Graphite as Graphite messages. Each metric value is an entry with a timestamp in a Graphite message. Below is an example of the graphite message payload sending attempted metrics values of a pipeline that defines a counter metric and a distribution metric.
  
  

Now you know about Beam metrics and their export !

Nexmark: benchmark and CI tool for Apache Beam

🕥 10 min. 

This blog post is about the subject of this talk I gave at the ApacheCon 2017. While the talk focuses on building Nexmark for Apache Beam, this article will focus on the use the Beam project has done of Nexmark since then. If you're interested in details about Nexmark, please refer to the talk.

What is Nexmark ?

Nexmark is originally a research paper of 2004 about benchmarking queries on a streaming system. In the context of Beam, what we call Nexmark is its adaptation in the form of Beam pipelines that run in both batch and in streaming modes. 

Nexmark is a simulation of an auction system. The synthetic data injected in the system is about Persons placing Auctions on Items and Persons placing Bids on these Auctions. 

Nexmark then interrogates the auction system for statistical data by running queries such as below.

The queries are written as Beam pipelines and these pipelines are meant to cover 100% of the Beam public API (including Beam SQL) to benchmark and detect regressions on all of Beam.

Query	Description	Use of Beam API
3	Who is selling in particular US states?	Join, State, Timer
5	Which auctions have seen the most bids in the last period?	Sliding Window, Combiners
6	What is the average selling price per seller for their last 10 closed auctions?	Global Window, Custom Combiner
7	What are the highest bids per period?	Fixed Windows, Side Input
9 *	Winning bids	Custom Window
11 *	How many bids did a user make in each session he was active?	Session Window, Triggering
12 *	How many bids does a user make within a fixed processing time limit?	Global Window, working in Processing Time

* Not in the original Nexmark research paper.

These queries are particularly suited for benchmark because they are representative of what a real user could request on a real auction system. They are also quite complex and leverage all the capabilities of Beam runners.

Throughout the execution of each pipeline, Nexmark gathers metrics (using Beam metrics system) such as the number of output results, event rate (input element generation rate) and obviously the query execution time.

What is the architecture of Nexmark?

Nexmark is a standalone executable which 

• creates a source that generates events (such as a Person creating an Auction or placing a Bid on an ongoing Auction)
• runs the queries depending on the workload selected by the user (ingestion rate, number of events generated, selected queries, etc...)
• monitors queries execution and collects their execution metrics.

Nexmark can run on all the supported Beam runners.

The user can chose to benchmark in batch mode or in streaming mode: Nexmark creates either a Beam BoundedSource to trigger batch mode in the runner or an UnboundedSource to trigger streaming mode in the runner.

Each query pipeline has this form:

1.Get PCollection<Event> as input

2.Apply ParDo + Filter to extract the object of interest: Bids, Auctions, Person

3.Apply Beam transforms that implement the query business logic: Filter, Count, GroupByKey, Window, etc...

4.Apply ParDo to output the final PCollection: collection of AuctionPrice, AuctionCount ...

How is Nexmark useful for Beam ?

    CI setup

With each commit on master, a jenkins script runs in a local in-JVM runner a Nexmark suite of queries with 100 000 generated events:

• on all the queries including Beam SQL versions
• on all the supported runners
• in both batch and streaming modes

The output of this benchmark is stored in Google BigQuery tables per query x runner x mode.

Each record of the table contains:

• the size of the output PCollection
• the execution time of the query

The results are then printed into graphs allowing to build performance dashboards and output size dashboards per runner as shown in the images below showing part of the flink dashboards:

    Performance assessment

Such dashboards allow to keep track of the performances of the different Beam runners:

        Detect performance regressions

The performance graphs allow to detect performance regressions. As query exercise a particular part of Beam API, it allows to narrow the field of search. Here query 6 uses Combiners so the regression showed below observed on 2018/10/05 on query 6 on the spark runner allowed to point to Combine translation in the spark runner and then fix it.

        Measure the impact of performance changes

When performance improvement tasks are run on a particular runner, we can check the impact on the graphs. Here there was a performance improvement on the spark runner on RDD caching that decreased the execution time of query 2 by more than 2. This change was committed just after 2018/11/21:

        Compare performances of the runners

Each query exercise a different part of the Beam API, so for each one we can compare the performances of the runner translations of that particular part of the API. The graph below, for example, compares the performances of the new Spark runner based on the Structured Streaming framework and the current Spark runner based on RDD/Dstream. The comparison is on query 8 which is a join (CoGroupByKey Beam transform).

    Functional assessment

There are also graphs that print the size of the output PCollection for each query, such graphs allow to check the functional behavior of the runners.

        Detect functional regressions

The output size graphs are useful to detect functional regressions: indeed, such a graph should be a straight line as, whatever the performance, a given query produces always the same results unless there is a bug: if the pipeline produces wrong data, then the size of the output PCollection will vary as in the example below. In this example, the flink runner produced wrong data for some time when executing query 12 in streaming mode. 

If a temporary failure raised an exception on a given query, it would lead to a gap in the graph.

        Verify compatibility

Beam provides the Capability matrix that shows the compatibility of the runners with the different part of the Beam SDK. The Nexmark output size graphs are also useful to verify and update the capability matrix. An unsupported API would raise an UnsupportedException. For example query 3 uses Beam state and timer API. This API is not yet supported on the new spark structured streaming runner. Thus, when the query runs, an exception is raised and the pipeline does not finish. As a consequence, no BigQuery table is created for this query and it leads to a hole in the dashboard as shown below.

    Release check

All these functions of Nexmark allow us to check candidate releases of Beam: we go through all the dashboards for all the runners in both batch and streaming modes seeking for performance or functional regressions. If any is found, the release process is interrupted and a new candidate release containing the fix is created. This way, the level of confidence in releases is pretty high because, as said, Nexmark aims to cover 100% of the Beam scope. 

Code callouts on blogger

🕥 3 min.

Why a code callout component ?

I spent quite some time to make my previous article readable. This article contains a long code extract which is quite unusual because it deals with java code generation. It requires a lot of explanations. Thus, I searched for a code callout component that is compatible with github gists.

Code callout component

I came by this component by Dave Leeds. It is brillant !
It allows to put links in the text to highlight lines in a gist snippet or other format snippets. The links are well integrated in your template comparing to some other tools that generate both the text and the snippet. See the extract of the article below

It highlights lines or separated group of lines. When you click the link to the line, it scrolls to the line in the gist and opens a popup with a personalized message and a button to scroll back to the text. See below

The separation between the link and the popup message allows to stay concise in the explanation text and give more details in the popup next to the code. See below

I will not explain how to use this component because it is already explained in the README file in the github repo but I will rather focus on the integration with blogger

Integration with blogger

After a quick change to the jquery selector, the component worked like a charm with the new github gist format. Where it is more complicated is with the integration with blogger. To work, the component requires several things:

• The jquery lib import: this one could be added to the HTML code of the blogger template. In the head section paste:

<script src='https://ajax.googleapis.com/ajax/libs/jquery/3.4.1/jquery.min.js'/>

• The code callout script import: the script is not hosted in any CDN, so you need to copy the minified code of the script available in dist directory of the repo to blogger. Adding this code to the HTML code of the template fails because blogger HTML editor refuses to take it. It might be a bug of the blogger editor. Adding it to a custom javascript gadget in the blogger layout also fails because the added gadget might not be rendered in mobile devices, so the script will not be called and the callout component will not work. The only way I found (might not be the best) is to paste the code of the script to a script tag at the end of the article that uses it.
• The call to the method that parses callout links in the article: right after the script of code callout (see previous step), paste 

<script>
    $(".callout-trigger").codeCallout({ profile: "gist" });
</script>

• Your links to code lines: see the README in the repo to learn what attributes you need to add to your links and to your gists to make it work.

Enjoy !

How to create a custom Spark Encoder in ... java

🕥 13 min

What is a Spark Encoder ?

An Encoder is a wrapper class that specifies how to serialize and deserialize data with the Spark Structured Streaming framework.

In this framework, datasets are typed and they are schema aware so every Spark transformation called on a dataset needs to provide an Encoder for its output type so that the framework can know how to serialize and deserialize its content. For example, the signature of the map transform is as this:

Why this blog post ?

Because Spark Structured Streaming Encoders are complex to write, have no java API and they are not very documented so I thought that what I learnt writing them could be useful to the community.

Catalyst

To understand Encoders we need to focus a bit on Catalyst. Spark Structured Streaming uses Catalyst optimizer to optimize execution plans. And Encoders are part of the Catalyst plan. Catalyst sees a pipeline as a tree made of TreeNodes. Here is the Catalyst workflow:

The entry point of Catalyst is either a SQL abstract syntax tree (AST) returned by the SQL parser if the user has used direct Spark SQL or a Dataframe object if the user has used the Dataframe/Dataset API. The result is an unresolved logical plan with unbound attributes and data types. During the analysis phase, Catalyst does catalog lookups and attributes mapping to fill these placeholders and ends up with a logical plan. This plan is then optimized by applying standard rule-based optimizations to end up with the optimized logical plan.  Then Catalyst uses rules to turn the optimized logical plan into one or more physical plans with physical operators (for example dataset.map) that match the Spark execution engine. It then selects a plan using a cost model that consists of applying cost evaluation rules to the physical plans.

When we come back to the subject is with the last phase: the code generation. Among the Treenodes types, there are Catalyst Expressions and custom Encoders are Catalayst Expressions called ExpressionEncoders. Catalyst Expressions contain java code strings that get composed to form the code AST of the pipeline which is then compiled to java bytecode using Janino compiler. This bytecode is the one that is executed when the pipeline runs. Let's see how to write ExpressionEncoders:

Custom Encoder (ExpressionEncoder)

To manage serialization in Spark Structured Streaming framework, you could use the Encoders available in the Encoders utility class. They can manage serialization of primitive types or beans with either java or kryo serialization. But let's say you want to write a custom Encoder because, for example, you develop a framework based on Spark and you want to allow your users to provide serialization code. As an example, let's take Apache Beam which lets the user define his serialization code in the Coder class.


Spark Encoders have no java api, so if you want to code in java because your code base is in java, you'll need to use some bindings. Maybe there are better ways, but as I'm not a scala developer, I could be unaware of them :)


As said, a custom Encoder is an ExpressionEncoder

ExpressionEncoder groups serializer and deserializer. You need to specify your serializer and deserializer as Expression instances. Let's see the serializer part:

Serializer

Things to point out in that code:

• It is an Encoder so it has no SQL representation (remember, Catalyst Expressions are broader than Encoders) so we  implement NonSQLExpression and it has one input and one output so we extend UnaryExpression.
• There are several methods to override:
• child(): the Encoder is part of Catalyst tree (see above) so we need to keep track of its child (the input of the UnaryExpression)
• doGenCode(): this method is responsible for producing the java code strings that will be compiled by Janino.
• datatype(): that is the type of the result of the evaluation of the Expression. In our case binary (because we are serializing an object to binary).
• the other overrides productElement(), productArity(), canEqual() and consequently equals() and hashcode() are due to the fact that there is no Java API of ExpressionEncoder so we need to implement Scala product specifics.

Code generation

Let's focus on the interesting method doGenCode():

This method generates the code in the comment line 32 in the form of java strings.

It returns an ExprCode, see line 52. This Block is constructed through string interpolation. This Block creation is managed line 49.

Now that the global architecture of this method is clearer, there is some pieces that could look weird:

• To access an object that is not a local variable part of the generated code block, we need to add a Catalyst reference to it, see line 24. In our case we reference the coder which contains the user provided serialization code.

• As said above the serializer is a UnaryExpression. This Expression has only one input Expression which is its child (see lines 25 and 32). We need to concatenate the child code and the actual serialization code so that everything can be compiled by Janino line 52.

Instantiate the serializer

Here is the code to create the EncodeUsingBeamCoder object (the serializer part of our ExpressionEncoder). To instantiate this class we need to pass it a reference to its child in the Catalyst tree (remember, the input Expression of the UnaryExpression). To obtain a reference to the Catalyst input Expression, we do like this: BoundReference(0, new ObjectType(clazz), true). There is only one input (because EncodeUsingBeamCoder is a UnaryExpression) so we get it at index 0 and we indicate the Datatype of the input and its nullability.

Spark physical plan

The serialization part of the physical Catalyst plan comes like this

SerializeFromObject [encodeusingbeamcoder(input[0, org.apache.beam.sdk.util.WindowedValue, true], WindowedValue$FullWindowedValueCoder(VarIntCoder,GlobalWindow$Coder)) AS binaryStructField#11]

WindowedValue is the type to serialize from  (the clazz)  and WindowedValue$FullWindowedValueCoder(VarIntCoder,GlobalWindow$Coder) is the Beam Coder provided in the pipeline.

Deserializer

The deserializer class DecodeUsingBeamCoder that you can see in the full code link below is completely symetric to the serializer class EncodeUsingBeamCoder. The only thing worth mentioning is its instanciation:

Instantiate the deserializer

Here is the code to create the DecodeUsingBeamCoder object (the deserializer part of our ExpressionEncoder). Here also, to instantiate this class, we need to pass it a reference to its child in the catalyst tree (remember, the input Expression of the UnaryExpression). To obtain a reference to the catalyst node, we do like this: new Cast(new GetColumnByOrdinal(0, BinaryType), BinaryType).

Here again we get the first input Expression at index 0 (cf UnaryExpression), it is of type BinaryType because we are deserializing bytes. And the Cast allows Spark to treat the Expression more efficiently as it allows Catalyst to treat it as binary.

Spark physical plan

The deserialization part of the physical Catalyst plan comes like this

DeserializeToObject decodeusingbeamcoder(cast(binaryStructField#4 as binary), org.apache.beam.sdk.util.WindowedValue, WindowedValue$FullWindowedValueCoder(VarIntCoder,GlobalWindow$Coder)), obj#6: org.apache.beam.sdk.util.WindowedValue

WindowedValue is the type to deserialize to (the clazz) and WindowedValue$FullWindowedValueCoder(VarIntCoder,GlobalWindow$Coder) is the Beam Coder provided in the pipeline

Performances

Even though Janino compiler is fast, compiling java strings to bytecode takes time. I measured a big performance gain when I reduced the size of the generated code and replaced it by as much compiled code as possible. This is why in full code link, encode() method in the serializer and decode() method in the deserializer are compiled code and not string code inside doGenCode() method. Another gain to this approach is to enable debugging of this compiled part of the code.

Full code link

EncoderHelpers class in the Apache Beam project

Understand Apache Beam runners: focus on the Spark runner

🕥 5 min.

Previously on Apache Beam runners 😀

In the previous article, we had a brief overview of what an Apache Beam runner is. This article will dig into more details.

The previous article introduced this very simple pipeline:

We saw that the Beam SDK translates this pipeline into a DAG representing the pipeline in the form of Beam transform nodes. Now let's see how a Beam runner translates this DAG. Let's say that the user choses Apache Spark as the target Big Data platform when he launches his pipeline.

The runner (at last)

The job of the runner is to translate the pipeline DAG into a native pipeline code for the targeted Big Data platform. It is this native code that will be executed by a Big Data cluster. If Spark is chosen, then the runner translates this DAG below into the Spark native pipeline code below. For the sake of simplicity the Spark pipeline is pseudo-code.

Composite and primitive transforms: the level of translation

In the previous article we talked about Beam transforms (primitive transforms and composite transforms). In the continuation of the blog, we will refer to "composite transform" as just "composite" and "primitive transform" as just "primitive".

The DAG above is the expanded DAG: on the left hand-side are the Beam transforms of the user pipeline. But among these transforms only Read is a primitive. The others are implemented by Beam SDK as composites of other primitives. Composites can also be made of composites themselves (like Count transform is made of Combine transform) but in the end they are always made of primitives (Pardo or GroupByKey). And these primitives are what the runner translates.

But in some cases, the runner can chose to translate at a composite level of the graph, not at a primitive level depending on the target Big Data technology capabilities. Indeed, if there is a direct correspondance of the Beam composite in the target API, the runner does not decompose the composite into its primitives and translates directly the composite. The green boxes in the DAG represent the level of translation. In our example, there is a direct equivalent of Beam Combine composite to a Spark Aggregator (agg in the Spark pipeline).

The translation itself

The translation occurs when the pipeline is run (pipeline.run() is executed). To translate the DAG, the runner visits the graph. All Beam runners work the same, only the target API changes with the chosen runner.

The first step is to detect the translation mode (batch or streaming) by searching for Beam BoundedSource (like Elasticsearch for example) or UnBoundedSource (like Kafka for example). Knowing the translation mode, the runner can chose

• the proper Spark DataSourceV2 to instantiate either implementing ReadSupport (batch) or MicroBatchReadSupport (streaming)
• the proper Spark action to execute on the output dataset either foreach (batch) or writeStream (streaming)

Then the DAG visit continues and each node is translated to the target native API : Read gets translated to a Spark DataSourceV2 that creates the input dataset, Pardo gets translated to a Spark flatmap that is applied to the input dataset and so on until the output dataset.

Pipeline run

When the visit of the DAG is done, the runner applies the action chosen above to the output dataset to run the pipeline. At this point the spark framework executes the resulting native Spark pipeline.