Queryable States in ApacheFlink – Part 2: Implementation

This is part 2 of the blog Queryable States in Apache Flink. In the previous blog, we saw how Apache Flink enabled Queryable States. In this part, we will create a Streaming Job with Queryable States and create a QueryClient to query the state. I assume that Flink is already installed and setup. If not you can check out my earlier blog post on installation here. I will be using a Tumbling window in this example, to read about Windows in Flink, please read this blog post.

All the code used in this blog post will be available on my GitHub.

Creating the Pipeline

Let us now create a streaming job with QueryableState. In this example, our input is climate log which is of the format country, state, temperature, humidity where country and state are Strings, temperature and humidity are Floats. We will first create case class to hold these logs.

We can then read the logs from a socket using

We will create a KeyedStream and apply a Tumbling TimeWindow of 10 seconds. This will cause the window to be evaluated each time it tumbles. In the apply function, we will do a simple aggregation to sum up all the values of temperatures and humidities seen in that window.

QueryableStateStream

Now we will create a Stream that is queryable. To do that, we need a StateDescriptor that describes the type of elements that are going to be stored in the stream. We will create a ReducingStateDescriptor that aggregates the values seen so far. The ReducingStateDescriptor takes three parameters, first parameter is the name, second is the reducing function that has to be applied when new elements are added to the state, and the third describes the type of values that are going to be stored in the state.

Once that is done, we call asQueryableState function to make the stream queryable and pass the state descriptor created.This is shown below.

Note the first parameter while calling the asQueryableState state function, this is the queryableStateName which is used for identifying the stream. This will be later used by the QueryClient while querying.

QueryClient

Now we will move on to the creating the QueryClient. The client is going to be a separate application that queries the state of an already running Streaming job. First thing that the client needs to know is how to connect to the JobManager (remember the diagram from the previous blog?), which can be configured as follows

Next we create an instance of QueryableStateClient and also Serializers for key and the value. The key serializer is used to create a serializedKey. The value serializer will be used later to deserialize the result returned back from the query. In the below example, we are asking the state to return the current running state value for the country “USA”.

Now we can query the state using the client. Pass the serializedKey, JobID and queryableStateName as parameters. JobID can be obtained either from the Flink UI or from the job submission log. Note that climatelog-stream parameter which should be same as the queryableStateName used during job submission.

The query returns a Future object which can be accessed as follows. If the query was successful, then we can use the valueSerializer to deserialize and read the result. In this case, the deserialized result is an instance of the ClimateLog case class.

To test the job, open a terminal and run netcat.

Now submit the job using flink command line interface

Now all that is left to do is send some sample messages through netcat, and run the QueryClient with the JobId and other parameters.

There are a few possible Exceptions that can occur at this point.

1) Actor not found

Make sure that your Flink cluster is up and running. Also you have to submit the Job through the command line, not from the IDE.

2) Job not found

This is an easy one, just make sure that JobId passed matches with that of the running job.

3) No KvStateLocation found

Make sure that the state name(climatelog-stream) in the client matches with the one that was used during job submission.

4) KvState does not hold any state for key/namespace

This means that the stream that you are tying to query does not have the key(in this example – “USA”) that you are looking for. Did the messages that were sent through netcat have the key that is being used in the query?

5) Could not deserialize value

Which indicates that something is wrong with the ValueSerializer. The easiest way to fix this is by going back to your Streaming Job code and making sure that you use the exact same TypeInformation in the client as used in the Job. e.g., using createTypeInformation[ClimateLog] instead of TypeInformation.of(new TypeHint[ClimateLog]() {}) can cause exception.

To summarize, we saw how Apache Flink enables querying it’s internal state and how we can develop a pipeline and query client to do so. Apart from Flink, Kafka also provides this feature.

That concludes the post and hope it was useful. Thanks for reading!