Streaming architecture

Some miscellaneous notes on streaming.

Streaming Semtantics are Difficult

"The assumption 'we receive all messages' is actually challenging for a streaming system (even for a batch processing system, though presumably the problem of late-arriving data is often simply ignored here). You never know whether you actually have received 'all messages' -- because of the possibility of late-arriving data. If you receive a late-arriving message, what do you want to happen? Re-process/re-sort 'all' the messages again (now including the late-arriving message), or ignore the late-arriving message (thus computing incorrect results)? In a sense, any such global ordering achieved by 'let's sort all of them' is either very costly or best effort." [StackOverflow, Michael G. Noll
Technologist, Office of the CTO at Confluent]

Kafka

"Brokers are not interchangeable, and clients will need to communicate directly with the broker that contains the lead replica of each partition they produce to or consume from. You can’t place all brokers behind a single load balancer address. You need a way to route messages to specific brokers." [Confluent Blog]

"But whatever you do, do not try to build an ESB around Kafka—it is an anti-pattern that will create inflexibility and unwanted dependencies." [Kai Waehner, Confluent]

Kafka as a Database

This comes from a talk by Martin Klepperman (Kafka as a Database).

He takes the canonical example used in SQL databases and extends it to Kafka. Let's send two parts of a financial transaction as a single message to Kafka. We then have Kafka separate it into two messages: one to a credit topic, one to a debit topic. We now have atomicity and durability but not isolation. If you think about it, this is the same as ReadCommitted. 

Note, if you crash half way between these two operations, you will need to dedupe or using Kafka's once-only semantics but you still have durability.

Example 2: imagine two users try to register the same user name. Can we enforce uniqueness? Yes, partition the topic on username and one will be before the other and all consumers will agree. We've achieved serializability (ie, an isolation level) and have lost parallelization.

Kafka Leaders and Followers - Gotcha

Jason Gustafson (Confluent) describes how Kafka logs can be truncated.

Say the leader writes locally some data but then dies before it can replicate that data to all of its ISRs. The remaining ISRs replicas nominate a new leader [via the "controller"] but it's an ISR that did not receive the data. Those ISRs that did receive it now have to remove it as they follow-the-leader but the leader never had it. Their logs are truncated to the high watermark.

Note that clients will not see this data as there is the notion of a watermark that is the highest commit ID across all ISRs. (Slow ISRs can be dropped from the set but this invariant still holds. The dropped ISRs can rejoin later when they have the requesite data).  The leader defines the high watermark. This piggybacks on the response to a fetch request. Consequently, the high watermark on the follower is not necessarily that on the leader.

Committed data is never lost.

Terminology

Reading documentation, references to out-of-bound data kept cropping up so I thought it worth making a note here:

"Out-of-band data is the data transferred through a stream that is independent from the main in-band data stream. An out-of-band data mechanism provides a conceptually independent channel, which allows any data sent via that mechanism to be kept separate from in-band data. 

"But using an out-of-band mechanism, the sending end can send the message to the receiving end out of band. The receiving end will be notified in some fashion of the arrival of out-of-band data, and it can read the out of band data and know that this is a message intended for it from the sending end, independent of the data from the data source." [Wikipedia]

Covariance and Precision Matrices

My aim is to write an advanced maths book that uses just pictures :) This is my attempt for the concepts of covariance and precision.

Example 1: Student IQs

Let's take a class of 100 students and take 10 measurements for each of them. This measurement data is Gaussian in its distribution. Then, our data looks like this:

Data Heatmap

[Python code lives here].

We can see each student's measurements roughly conforms to a value but there is a large variety over all the students.

Now, we build the covariance matrix where:

1. the expected value (mean) for a row is subtracted from each value in the vector
2. the resulting vector is multiplied with all the other vectors and this forms the index for our matrix
3. each cell is multiplied by the probability of that combination occuring. It's perfectly possible that this is zero in some cases but for our use case the distribution of which student and which measurement we might chose is uniform.

Note, regarding covariance as a measurement:

"Covariance is useful in some computations, but it is seldom reported as a summary statistic because it is hard to interpret. Among other problems, its units are the product of the units of X and Y. So the covariance of weight and height might be in units of kilogram-meters, which doesn’t mean much" - Think Stats

Anyway, our covariance matrix looks like this:

Covariance matrix heatmap

It should be immediately obvious that 

1. each row has a strong covariance with itself (the diagonal)
2. that the matrix is symmetric (since the covariance formula is symmetric) 
3. and that all other cells than the diagonal are just noise.

The inverse of the covariance matrix is called the precision matrix. It looks like this:

Precision matrix heatmap

This represents the conditional probability of one student's measurements on another and not surprisingly, in our particular use case, the matrix is full of zeroes (plus some noise). In other words, the measurements for all students are independent. And why shouldn't they be? Given the measurements for student X, what does that have to do with the measurements for student Y? This might not be true for other datasets as we'll see in the next example. 

Example 2: A Stock Price

To illustrate this, I've written some Python code based on this excellent post from Prof Matthew Stephens. He asks us to imagine a random walk where on each step we add the value from a Gaussian distribution. The output might look like this:

Random walk

Note that just adding Gaussians qualifies as a Markov Chain as the next value only depends on the previous. That is:

X₁ = N(σ², μ_i)

X₂ = X₁ + N(σ², μ_i)

X₃ = X₂ + N(σ², μ_i) 

...

X_n = X_n-1 + N(σ², μ_i)

We rewrite this so that our random variable, Z, is depicted as a vector of values generated from N(σ², μ_i) and note that a random variable is neither random nor a variable in the programming sense of the words. It's values may have been drawn from a random distribution, but once we have them they are fixed as you can see from our Python code. 

Anyway, once we've done this, we can represent the above set of equations as matrices:

X = A Z

where A is just an n x n matrix where the lower left values are 1 and the others are 0 (try it).

The covariance matrix of A tells us how each of the elements are related to each other and looks like this:

Covariance Matrix of the Random Walk

Unsurprisingly, values are similar to their neighbours (the diaganol) and less similar to those further away (bottom left and top right corners). We'd expect that with a random walk.

[Just a note on covariance matrices: each row on n elements is centred on its mean. The resulting matrix is then multiplied by its transpose and divided by n - 1 (the -1 is the Bessel correction that comes from the fact we're sampling rather than using the entire population). But this depends on a certain interpretation of what a row is. If the row is taken from the same distribution, happy days.]

The precision matrix looks like this (zoomed into the top 10 x 10 sub matrix):

Precision Matrix of a Markov Chain

and this gives us an intuitive feeling for what the precision matrix is all about "The key property of the precision matrix is that its zeros tell you about conditional independence. Specifically [its values are] 0 if and only if X_i and X_j are conditionally independent given all other coordinates of X." [here] And this is what we see here. Each point is only dependent on its immediate neighbour - the very definition of a Markov process.

Optimizing Spark/DataBricks in Azure

We have about 1.5TB of parquet data in Gen 2 storage on Azure and we're trying to read it with a DataBricks cluster of 8 nodes each with 56gb and 16 cores. All I'm trying to do is some basic select('aColumn).distinct and where('anotherColumn < X).count()s.

Surprisingly, this is running like a dog: the first query takes about 50 minutes and the second about 18. Tuning is an empirical science but this felt wrong as 1.5TB of data is not that much these days.

Ganglia Metrics

Ganglia showed that the cluster was running at high capacity although the network usage seemed low. If this is showing the data being pulled from Gen2 into Spark workers, at 300mb/s it would take about an hour to pull it all out of storage. Indeed, this is roughly how long it took. 

However, when I took thread dumps, the executor threads were not waiting on IO. Typically, they were in Spark's CompressibleColumnBuilder (which, according to the docs, "builds optionally compressed byte buffer for a column"). So, it seemed that the cluster really was CPU-bound on parsing the files and garbage collection - which was bad but not horrendous:

Spark GC logs

Calling cache() on the data didn't seem to make queries much faster. Looking at the thread dumps again showed the threads spending a lot of time in CompressibleColumnBuilder for the initial select distinct and in java.io.FileInputStream for the subsequent counts. 

Remembering that  persisting with a storage level of MEMORY_AND_DISK_SER had given me a quick win in the past, I thought I'd try it again but with no success.

I tried enabling the Delta Cache. This is supposed to cache the Gen2 data in the cluster:

Storage Tab

Compare this to the storage tab when we're not using Delta Cache and you'll see there won't be any data in the "Parquet IO Cache". (Note that the page says "RDDs" when we're only using the Dataset API. This is standard). Anyway, enabling Delta Cache seemed to make no difference at all.

Using a "Delta Cache Accelerated" cluster actually made the initial distinct.count() take much longer (1.47 hours) with only marginally reduced time for the next query (16.8 minutes).

Disk is incredibly slow compared to memory. Ultimately, I was trying to process too much data with too little RAM and there's only so far tuning will get you before your metrics plateau. So, I made a cluster of 25 nodes each with 28gb of RAM and 8 cores. This did help but not as much as I was expecting. The distinct.count() now took 38 minutes and the where(...).count() 12 minutes. This is between a 25%-33% improvement but still seems a long time in absolute terms.

I want my Data Locality 

The problem appears to be that all the data is pulled to the executors even if we're only interested in just one column [see StackOverflow]. Evidence for this came from another look at Ganglia:

Ganglia Metrics during second query

This was taken after the distinct.count() and during the where(...).count() query. Note that there is no more network activity, suggesting the first query cached the data.

"In Azure really fast networks compensate for having data and compute separated." [ibid] Indeed, the threads seemed to spend all their time on the first call in code to parse the Parquet rather than blocking on IO. But you can't get over the fact that my two queries are doing much the same thing (no joins) but the first takes three times longer than the second. To demonstrate, instead of two different queries, I ran the distinct.count() query twice. The first time took 42 minutes, the second 15. So, it appears that pulling 1.5TB of data from Gen2 into Spark takes about 25 minutes.

Azure is not the only cloud offering that implements this architectural trick. GCP takes all the data out of storage and puts it into an ephemeral HDFS cluster.

Solution

A new day, a new dawn. Let's start the cluster again and go from the beginning. This time, I won't explicitly enable IO cache and I won't even call .cache() on my DataFrame. To my amazement, the same query ran in about 30 seconds today! It seemed that this .cache() call was the root of my problems as re-introducing it returned us to a 50 minute wait. This is pretty much the opposite of what you'd expect with an on-prem Hadoop/YARN cluster.

A waste of two days

 This post is just for reference when people ask me what can possibly go wrong with a simple process. In this case, the solution was trivial but the problem was enormous.

Basically, I had some code for which decrpytion was central. I had all my tests run on each check-in that used Bouncy Castle to decrypt data encrypted by gnugpg. This was also tested in lower environments but the upper environments were closed to me. Only DevOps could go there. And, lo-and-behold! decryption failed there with exactly the same artifact as was tested elsewhere.

The error was "checksum mismatch at 0 of 20". A quick Google turned up this StackOverflow post that indicated there was a problem with the pass phrase (a means of protecting the private key). I looked at the file in which the passphrase lived using:

Playing with hexdump [SO] I ran something like:

hexdump -e'"%07.8_ax  " 8/1 "%03d "  " |\n"' passphrase.txt
00000000  049 050 051 -62 -93 052 053 054 |
00000008  055 010                         |

Well, firstly there appears to be a line return (010) in there but this is a read herring as if you use the command to get this into KeyVault as outlined in a previous post (that is, by cating it) then the new line is removed. This was serendipitous as the gpg man pages say:

       --passphrase-file file

              Read the passphrase from file file. Only the first line will be read from file file. 

If we'd used the proper az command line (using the -f switch to indicate a file rather than cating it), Key Vault would have contained the newline. 

So, after wasting some time on that, I next looked at why my JVM was saying the pass phrase was 21 characters when I could only see 20. Checking each byte of the pass phrase, it became clear that the character "£" (that is, the pound sterling sign) was taking up two bytes. 

Fair enough, I thought. It's not ASCII but I can encrypt and decrypt fine with this pass phrase using gpg. So, why was my code failing?

And this is the kick in the pants. "Unless otherwise specified, the character set for text is the UTF-8 [RFC3629] encoding of Unicode [ISO10646]." (OpenGPG format RFC) But there is a UTF-8 enconding for the pound sterling symbol, so what gives? Well, different implementation are at liberty to map characters to UTF-8 as they see fit, it seems:

"Please note that OpenPGP defines text to be in UTF-8.  An implementation will get best results by translating into and out of UTF-8.  However, there are many instances where this is easier said than done.  Also, there are communities of users who have no need for UTF-8 because they are all happy with a character set like ISO Latin-5 or a Japanese character set.  In such instances, an implementation MAY override the UTF-8 default by using this header key.  An implementation MAY implement this key and any translations it cares to; an implementation MAY ignore it and assume all text is UTF-8." [ibid]

So there is some ambiguity how to handle these characters and I don't appear to be the only one to have fallen into this trap as the Bouncy Castle mailing lists suggest. And it's not just Bouncy Castle but Open Keychain also appears to have this problem.

Ambiguous specificationas are the root of all evil.

Gaussian Processes - part 1

Gaussian Processes is a massive topic and these are just notes I made from learning about them.

The gist

It's all Bayesian.

Information Theory, Inference, and Learning Algorithms (MacKay)

If you squint, this is just Bayes' formula but where we're finding the probability for a formula, y(x) over the targets, t. and the data, X.

We assume we can model this problem by assigning Gaussians to all states.

Now, a multivariate Gaussian (sometimes called MVN for multivariate normal) looks like:

 MVN (ITILA, MacKay)

ITILA says "where x is the mean of the distribution" and that's the case when our expection is the mean. But if we applied this to a linear regression then x wouldn't be our mean but our linear equation (by definition of what is our expectation). Furthermore, why choose linear regression? Let's use a different function altogether! "This is known as basis function expansion" [MLPP, Murphy, p20].

The matrix, A, in the MVN equation above is just the inverse of the covariance matrix, called the precision or concentration matrix. Just as a reminder, covariance is defined as:

cov[X, Y] = E[X - E[X])(Y - E[Y]) = E[XY] - E[X] E[Y]

and so the covariance matrix is defined as:

cov[x] = E[(x - E[x])(x - E[x]^T]

Example  

This example is from examples for Figaro. Here, 

y(x) = x² + N(0, 1) 

where N is the normal distribution, x is an integer in the range (1, 10) and we want to find out if given, say x=7.5, what would y most likely be?

To do this, we introduce a covariance function:

exp ^{-γ * (x₁ - x₂)2}

where γ is just a constant and the xs are pairwise actual values. The choice of function just makes the maths easier but suffice to say it penalizes large differences.

Now, given our covariance function, we can apply it to all pair combinations to make the covariance matrix (represented as a heat map here):

Matrix from the covariance function

then we invert it (after we add some noise to the diaganol to ensure it's not ill-conditioned):

Inverted matrix of the covariance function

then element-wise multiply it with the observed y(x) values (ie, all those corresponding to x in the range 1 to 10):

Alpha

This completes the training process.

Now, all the Figaro example does is plug these tensors into pre-packaged formulas for the expected value (μ_*) and covariance matrix (Σ_*) of the derived Gaussian. These can be found in Kevin Murphy's excellent "Machine Learning: a Probabilistic Perspective", equations 15.8 and 15.9: 

μ_* = μ(X_*) + K_*^TK^-1(f - μ(X))
Σ_* = K_** - K_*^TK^-1K_* 

where

X is the observed data

f is the mapping from the known inputs (X) to outputs

X_* is the data for which we wish to predict f 

K is the observed data that has gone through the kernel function/covariance matrix and is a function of  (X, X)

K_* is a mix of the data X and X_* that has gone through the kernel function/covariance matrix and is a function of  (X, X_*)

K_** is the data X_* that has gone through the kernel function/covariance matrix and is a function of  (X_*, X_*)

The derivation of these equations is beyond me, as is why the Figaro demo uses 1000 iterations when I get similar results with 1 or 10 000.

Next time. In the meantime, it appears that Murphy's book is going to be available online in a year's time and the code is already being translated to Python and can be found at GitHub here.

Further reading 

Gaussian Processes for Dummies [Kat Bailey's Blog]

The Model Complexity Myth [Jake VanderPlas's Blog]

Spark Gotchas

Some miscellaneous notes on Spark I've been making the last few weeks.

Triggers and OutputModes when Aggregating

I was experimenting with aggregating a query that reads from Kafka and writes Parquet to HDFS.

The interplay between watermarking, Trigger and OutputMode look a bit like this:

• Queries that aggregate with OutputMode.Append yet with no watermark will always throw "Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark;;" irrespective of the Trigger. 
• Queries that aggregate with OutputMode.Complete  with "Data source parquet does not support Complete output mode;". Even text files will barf with "Data source text does not support Complete output mode;".
• Queries that aggregatie with OutputMode.Append but with a watermark behave like this:

Trigger	Behaviour
Continuous	Unknown type of trigger: ContinuousTrigger(5000) [1]  or Continuous processing does not support EventTimeWatermark operations.;;
ProcessingTime	No errors but last batch is not pulled from Kafka
none ("When no trigger is specified, Structured Streamin stats the processing of a new batch as soon as the previous one is finished." - Stream Processing with Apache Spark)	No errors but last batch is not pulled from Kafka
Once	No errors but nothing is pulled from Kafka. Running the query twice with awaitTermination pulls some elements.

The reason Spark leaves a little something on Kakfa is (apparently) for efficiency. There appear to be a pull request for making Spark consume the last bit from Kafka but in its associated Jira somebody mentions my StackOverflow question that talks about Spark's behaviour being unchanged.

So perplexed was I that I asked the Spark mailing lists. The answer from Spark committer, Jungtaek Lim, was illuminating:

"Unfortunately there's no mechanism on SSS to move forward only watermark without actual input, so if you want to test some behavior on OutputMode.Append you would need to add a dummy record to move watermark forward."

This doesn't seem to be because we're writing Parquet. Text files will also behave this way.

Defragging Question

Spark will create a new file per executor thread per ProcessTrigger time (as this test code demonstrates). This can lead to a lot of files that could ultimately choke your Name Node if you're persisting to HDFS. This may necessitate a roll-your-own fix [SO].

If you're using DataBricks, you can do this "bin-packing" by running the OPTIMIZE command. Unfortunately, I could not see the code to do this inside the Delta library DataBricks have open sourced. (I discuss other ways of dealing with this problem without Delta in this blog post).

Since our use case at work is such that we only want the output of stream processing once a day, we opted for Trigger.Once. Since we're not aggregating, all of the data is pulled and processed. Since this is streaming in a very loose sense of the word (it's more batch processing that gets its data in one short, sharp stream), we avoided having large numbers of small files.

Exceptions in processing DataFrames

Spark can optionally have a "best efforts" approach to exceptions rather than failing the whole process.

spark.range(10000).repartition(7).map { i =>
  if (i == 9999) { Thread.sleep(5000); throw new RuntimeException("oops!") }
  else i
}.write.option("mapreduce.fileoutputcommitter.algorithm.version", ALGO_VERSION).mode("append").parquet(s"/tmp/test_algo_$ALGO_VERSION")

When ALGO_VERSION=1, nothing is written to HDFS. On the other hand, when ALGO_VERSION=2, the partition in which the exception occurred is not written but all the other ones are.

see https://databricks.com/blog/2017/05/31/transactional-writes-cloud-storage.html

DataFrame types

You can see the schema as JSON with:

spark.read.parquet("THE_FILE").schema.json

The type "timestamp" would map the data to a java.sql.Timestamp (the equivalent of a SQL TIMESTAMP type). This Java type is a subclass of java.util.Date.

The type "date" would map to a java.sql.Date (the equivalent of a SQL DATE type). This is also a subclass of java.util.Date.

See this StackOverflow page  for descriptions of the SQL types.

Trying to read from a Parquet file that uses "date" and coercing it to a "timestamp" will result in:

Caused by: java.lang.UnsupportedOperationException: org.apache.parquet.column.values.dictionary.PlainValuesDictionary$PlainIntegerDictionary

Idempotency

DataStreamWriter.foreachBatch is experimental and "is supported only the in the micro-batch execution modes (that is, when the trigger is not continuous)". It provides a batchId that "can be used deduplicate and transactionally write the output (that is, the provided Dataset) to external systems. The output Dataset is guaranteed to exactly same for the same batchId (assuming all operations are deterministic in the query" [Spark docs].

[1] "You can set the optional Continuous Trigger in queries that satisfy the following conditions: Read from supported sources like Kafka and write to supported sinks like Kafka, memory, console." SO quoting DataBricks.

Azure

The Azure web GUI is quite immature. For instance, if you install the Azure Storage Explorer (Windows only) it doesn't show timestamps of files. Fortunately, a lot (everything?) can be done from the command line. This, for instance, mounts a SMB drive in the cloud on my local Linux box where it can be treated as any other directory:

sudo mount -t cifs //XXX.file.core.windows.net/DIRECTORY /mnt/DIRECTORY -o vers=3.0,username=USERNAME,password=GET_THIS_FROM_THE_WEB_GUI,dir_mode=0777,file_mode=0777,serverino 

Also, if you want to put a multi-line value into Microsoft's Key Vault, you'll find you can't do it in the web GUI. You need to put the text with line returns into YOUR_FILE and use:

az keyvault secret set --name YOUR_KEY --vault-name VAULT_NAME --value "`cat YOUR_FILE`"

Docker and K8s in the Azure cloud

First, tag your image with something like:

docker tag 8d2be7e5d4eb XXX.azurecr.io/YYY:1.0

where XXX is your image repository subdomain in Azure and YYY is the name of the artifact. Login with:

az acr login -n XXX

and now you can push your artifact into the Azure infrastructure:

docker push XXX.azurecr.io/YYY

(You might need to run az acr login -n XXX first) 

Let's check it works:

kubectl run -i --tty --attach ARBITRARY_NAME --image=XXX.azurecr.io/YYY:1.0  --command -- /bin/bash

and behold, we are on the CLI of a remote container in the Azure cloud.

But don't forget to clean up after ourselves with:

kubectl delete deployment ARBITRARY_NAME

Network Speeds

By having your image pushed to K8s, you can run your code in Azure as easily as your laptop. The big benefit is network speeds. In my case, I was decrypting an RSA encoded file taken from BLOB storage at about 1mb/s on my (well specced) laptop but exactly the same code was easily managing 10mb/s in the Azure cloud. (Yes, I know that using asymmetric ciphers for large files is not efficient [SO] but this was imposed on us by our client). By using jstack, I could see that the threads on my laptop were spending most of their time in IO not Bouncy Castle.