%md 
# Streaming TDigest with flatMapGroupsWithState
by [Benny Avelin](https://www.linkedin.com/in/benny-avelin-460b99121/) and [Håkan Persson](https://www.linkedin.com/in/håkan-persson-064b763/)

The idea with this sketch is to demonstrate how we can have a running t-Digest in a streaming context.

## Arbitrary stateful aggregations in streaming
We have two stateful operations, the first is mapGroupsWithState and flatmapGroupsWithState. The Databricks blog have a relatively good explanation of the two operations in their blogpost https://databricks.com/blog/2017/10/17/arbitrary-stateful-processing-in-apache-sparks-structured-streaming.html.
However the concept is maybe not so easy to understand so I will try to give a simple explanation of what is going on with these two aggregations.

### Structured streaming
For the purpose of this sketch we only need to know that new data will arrive as a batch, if we instead of a streaming dataframe just apply the aggregations on a dataframe then the entirety of the data will be in a single batch.

### A running state
The way both mapGroupsWithState and flatMapGroupsWithState works is that we start with a key-value grouped datasets, when new data arrives it will be split into the groups prescribed by the key and each key will get a batch of data. The main important idea to realize is that for each key we have a running state, and there is no prerestriction to witch keys are ok and not so the number of keys can grow/shrink or whatever. If a new key appears, the first step in both mapGroupsWithState and flatmap... is to initialize a zero state before processing the first batch for this key, the next time a key appears it will have remembered the previous state and we can use the previous state and the added batch of data to compute the next state. What can a state be? Well an object of some class that has been predescribed, the simplest would be a running max/min/mean but also as we will see in this sketch a t-digest.

### flatmapGroupsWithState vs mapGroupsWithState
The simple difference between these two can be infered from the name, but let us go into detail. If we are only interested in an aggregated "value" (could be a case class) from each key we should use mapGroupsWithState, however there are some interesting caveats with using mapGroupsWithState. For instance certain update-modes are not allowed as well as further aggregations are not allowed. flatmap... on the other hand can output any number of rows, allows more output-modes and allows for further aggregations, see the Structured Streaming programming guide.

<table>
  <tr>
    <td>Query type</td><td>Output mode</td><td>Operations allowed</td>
  </tr>
  <tr>
    <td>mapGroupsWithState</td><td>Update</td><td>Aggregations not allowed</td>
  </tr>
  <tr>
    <td>flatMapGroupsWithState</td><td>Append</td><td>Aggregations allowed after</td>
  </tr>
  <tr>
    <td>flatMapGroupsWithState</td><td>Update</td><td>Aggregations not allowed</td>
  </tr>
</table>

%md # Some streaming input
We need to have a streaming source for our example, this can be done in a number of ways. Probably there is some nice way to do this simply but the few methods I know to generate test-samples is to get a running loop that writes files with data, so that each time a new file arrives Spark will consider it as an update and load it as a batch. We have provided some code to generate points sampled from a normal distribution with anomalies added as another normal distribution.

%md # The /tmp folder
Databricks community edition has a file-number limit to 10000 and after running databricks for a while one will start to notice that things fail, and skimming the stacktrace of the failure we realize that we have reached said limit. Deleting files that one has created does not seem to solve the issue, well... this is because the /tmp folder counts into the limit and this is not cleared nearly as often as would be good for our work. Therefore we just clear it before starting our job...

ps. If you have not cleared the tmp folder before this might take some time actually. ds.

%md # AWS eventually consistent
The AWS distributed filesystem is eventually consistent, this can mean for instance that a file just created will not be possible to read and if we are unlucky the following code will fail to run. 

%md # States and rows
To begin describing the code below, let us first look at what will be our running State. The `isarnproject` sketches packs the TDigest class into a TDigestSQL case class and provides encoders for this to be allowed in a Dataframe, therefore we can capitalize on this and use TDigestSQL as our running state (to be precise it is the TDigest wrapped by TDigestSQL that is the state but whatever.). The next thing to worry about is how should we output and what should we output? This example shows how to embed in a single row, the TDigest, the threshold value that comes from `cdfInverse(0.99)` and the actual data that is above the threshold. To do this we create a case class which will be the template for our row, in the code below it is called `TdigAndAnomaly`.

## updateAcrossBatch
This is our main update-function that we send as a parameter to flatmapGroupsWithState. 

* It takes as first input the key-value, which we will not care about in this example and is just a dummy for us. 
* The second input is the `inputs : Iterator[timedScoreCC]`, this is an iterator over the batch of data that we have recieved. This is the type-safe version, i.e. we know that we have a `Dataset[timedScoreCC]`, if we dont and we instead have a `DataFrame = Dataset[Row]`, we have to use `inputs : Iterator[Row]`, and we have to extract the columns of interest cast into the appropriate types. 
* The third input is the running state variable, this is always wrapped in a `GroupState` wrapper class, i.e. since `TDigestSQL` was our state we need to have `GroupState[TDigestSQL]` as `oldstate`.
* Lastly we have the output, which is an iterator of the case class chosen as outputrow, in our case this is `Iterator[TdigAndAnomaly]`

Each time a batch gets processed, the batch data is in the `inputs` variable. We first make sure that the state is either the previous state (if it exists) or we set it to a zero state. Then we simply process the batch one datapoint at the time, and each time calling updateTDIG, which simply updates the state with the new data point (tDigest add point). Once we have added all the points to the t-Digest, we can compute the updated value of `threshold` using `cdfInverse(0.99)`, after that we simply filter the batch to obtain an iterator of the anomalies.

### GroupStateTimeout
This is an interesting variable that you really should look into if you wish to understand structured streaming. Essentially it is the whole point of messing around with the structured streaming framework, see the programming guide.