Notes on Comonads

This excellent post by Eli Jordan can be summarized as comonads offer counit and cojoin, the opposites of on unit and join (or flatten as Cats calls it) monads. 

Why this is useful is that it can model a use case where there is the notion of focus. To demonstrate this, Jordan has created an implementation of Conway's Game of Life (code). Here, the focus is a cell which we determine is either on or off given its neighbours. (Another example of how comonads use focus can be found here where we traverse a graph, node by node).

The case class on which we will call counit and cojoin is an abstraction, Store[S, A], but for the Game of Life example, it is the grid on which the changing cells live. Notably, Store is instantiated with two arguments, 

1. lookup: S => A
2. index: S  

Specifically for our game, these represent:

1. a mapping from a co-ordinate (S) to a Boolean (A) representing whether the corresponding cell is turned on or off.
2. the co-ordinates of the cell that has the focus. That is, S for us is of type (Int, Int).

For each step in our game, we ask our grid (Store) what the neighbourhood is like (lookup) for its focused cell (index). 

We do this by calling its Cats coflatMap function (that we added via a type class) that takes a grid and returns a Boolean. 

This is where it gets interesting. You'll notice that Store has the potential for a partially applied constructor. That is, it looks like:

case class Store[S, A](lookup: S => A)(val index: S) { ...

So, if we instantiate Store with just a S => A, we actually have a S => Store[S, A].

In our Game of Life, this means that for the coflatMap branch of the code, the grid is not actually a mapping to whether the cell is turned on or off (F[Boolean]) but to the grid of the previous iteration on which we will apply Conway's rules to derive Boolean on/off  mappings. That is, it's our cojoin of type of F[F[Boolean]] that needs to be flattened.

Memoization

Recalculating Conway's rules recursively is expensive. Without memoization, you'll see deep stacks. Using jconsole, I saw quite a few:

- eliminated <owner is scalar replaced> (a Store) at Store.counit(Life.scala:69)

which apparently "represents a redundant lock that has been removed in the bytecode." [SO]. This is because the code on each lookup is trying to recalculate the grid from scatch. Memoization prevents this by caching the grid of the last iteration.

What triggers this recalculation is the rendering of the 20x20 grid. This grid obviously leads to 400 evaluations. But the grid (Store) of each iteration has a reference its predecessor's 400 on which it must calculate Conway's rules. Not only do we not want to recalculate that grid from the game's beginning, we don't want to perform a further 3600 (=400 * 8 neighbours) calculation for cells we've already evaluated - hence the caching.

DR in an Azure Cloud Ecosystem

In our topology, we have a Databricks Spark Structured Streaming job reading from an HDInsights Kafka cluster that is VNet injected into our subscription. So, disaster recovery has to take into account two things:

1. the Kafka cluster
2. the blob storage where the data is landed by SSS.

Kafka

This Microsoft document outlines the usual way in securing your data in Kafka (high replication factors for disk writes; high insync replicas for in-memory writes; high acknowledgement factor etc). In addition, an HDInsight cluster are backed by managed disks that provide "three replicas of your data" each witihin their own availability zone that's "equipped with independent power, cooling, and networking" [Azure docs].

So, within a region, things look peachy. Now, how do we get these Kafka messages replicating across region? The HDInisghts documentation suggests using Apache MirrorMaker but note one critical thing it says:

"Mirroring should not be considered as a means to achieve fault-tolerance. The offset to items within a topic are different between the primary and secondary clusters, so clients cannot use the two interchangeably."

This is worrying. Indeed, there is a KIP to make MirrorMaker 2 fix this problem and others like it (like differences in partitions within topics of the same name; messages entering infinite loops etc). Confluent is pushing its Replicator that (it claims) is a more complete solution (there's a Docker trial for it here). And, there is Brooklin, but Azure says this would be self-managed.

Spark and Blobs

At the moment, all the data goes into one region. The business is aware that in the event of, say, a terrorist attack of the data centres, data will be lost. But even the infrastructure guys have the data being replicated from one region to another, note this caveat in the Azure documentation (emphasis mine):

"Because data is written asynchronously from the primary region to the secondary region, there is always a delay before a write to the primary region is copied to the secondary region. If the primary region becomes unavailable, the most recent writes may not yet have been copied to the secondary region."

But let's say that nothing so dramatic happens. Let's assume our data is there once we get our region back on its feet. In the meantime, what has been landed by SSS in the backup region is incompatible with what came before. This is because Spark stores its Kafka offsets in a folder in Gen2. It's just as well Spark is not writing to the directory that the erstwhile live region was using. If we had been writing to a directory that was common to both regions, some finagling would have to be done as we points the Spark job at another directory, effecting the RTO if not RPO.

Aside: A disaster of a different kind

In the old days of on-prem Hadoop clusters, you might see a problem where too many files were created. The consequence would be the Name Node goes down. Sometimes deciphering the Azure documentation is hard but this link  says the "Maximum number of blob containers, blobs, file shares, tables, queues, entities, or messages per storage account" has "No limit". 

Hopefully (caveat: I have not tested this in the wild) this problem has gone away.

Cache vs Delta Cache

Continuing the investigation into a lack of cache coherency in Databricks in my previous post,  I've raised the issue with the people at Azure.

In the document a representative from Microsoft pointed me to, there are two caches at play: Delta Cache and Apache Spark cache. "The Delta cache accelerates data reads by creating copies of remote files in nodes’ local storage" and is enabled by default on certain clusters. In effect, you get it for free when you use Databricks 

The Apache Spark cache is familiar one that is invoked by calling Dataset.cache().

My first complaint is that these two caches behave inconsistently. In the Delta Cache, updates are immediately available. But if we use the Apache Spark cache in conjunction with the Databrick's Delta Lake (not to be confused with the orthogonal Delta Cache), the data is frozen in time.

Now, in the Spark world, Datasets are strictly immutable. But Delta Lake "provides ACID transactions ... on top of your existing data lake and is fully compatible with Apache Spark APIs." Well, once I've invoked .cache(), and update the Delta Lake in the standard way:

    df.write
      .format("delta")
      .mode("overwrite")
      .option("replaceWhere", CONDITION_STRING)
      .save(HDFS_DIRECTORY)

(working example is here) then I cannot see my own updates!

What is, perhaps, even more bizarre can be seen when I try to read the data afresh. A second, totally independent call (but using the same Spark session) to read the data like this:

session.read.format("delta").load(dir)

cannot see the data either! And just to show you that there is nothing up my sleeve, running a completely separate process (code here) to read the data shows that the change has indeed been persisted to HDFS. This appears to be because Spark caches the data plus the Delta Lake meta data too and cares not for Databricks' new semantics.

This brings me to my next complaint - the leaky abstraction. Databricks is trying to retrofit ACID transactions on an API that was built with immutability at its core. The claim that it's "fully compatible with Apache Spark APIs" seems not enitrely true.

I have a meeting scheduled with the MS representative who is currently of the opinion that we should just never call .cache(). On Azure Databrics, this does not seem too bad as the Delta Cache seems pretty fast. It sucks for anybody using Delta Lake on just HDFS.

Summary

If you call .cache() on a Spark Dataset while using Databricks Delta Lake format you will not be able to:

• see updates from any other Spark session
• see your own updates to this data
• see the true data even if you re-read it from the underlying data store

unlesss you unpersist the Dataset.

[Addendum: I spoke to Databricks engineer, Sandeep Katta, on 17 December 2020 and he agreed the documentation is misleading. He says he'll generate a docs PR and submit it to the relevant team to make it clear that one should not use .cache when using Delta Lake]

Cache in the Azure

Gotcha!

What surprised me with the Azure/DataBricks/Spark architecture was a lack of cache coherence. Data is pulled from Azure's Gen2 and is stored closer to the Spark executors upon calling cache(). However, if another cluster updates that data, don't expect the first cluster to see it.

In my case, I was using DataBrick's Delta format to overwrite data as outlined here in the Delta Lake docs. But first, a business requirement demanded that if we're overwriting anything, an override flag must be set. However, because of a lack of cache coherence, this check indicated that there was nothing to overwrite even if another cluster had written that partition. This caused my code to overwrite what was there irrespective of the flag! This significantly changes the semantics of my Big Data application.

Furthermore, any writes after cache() has been called are not reflected in the original Gen2 storage in Azure...

"The Delta cache automatically detects when data files are created or deleted and updates its content accordingly. You can write, modify, and delete table data with no need to explicitly invalidate cached data." [DataBricks Delta Cache docs] This doesn't appear to be the case, at least if it's referring to the original file(s).

"Multiple writers across multiple clusters can simultaneously modify a table partition and see a consistent snapshot view of the table and there will be a serial order for these writes" [DataBrick Delta docs] Again, this might be true but not after cache() has been called, it appears.

The only interesting piece of architecture is that the Azure Gen2 storage was mounted using:

dbutils.fs.mount(  
  source       = s"wasbs://${dst}@${account}.blob.core.windows.net/",  
  mountPoint   = x,  
  extraConfigs = Map(s"fs.azure.account.key.$account.blob.core.windows.net" -> dbutils.secrets.get(scope = "discope", key = key))

Investigation continue but a friend at a completely different company that uses Databricks confirmed it's not me to whom this is happening. He too tried to replace a partition with a .option("replaceWhere=...") and although it appeared to work "locally" it did not change the underlying file store changed the underlying data store but the problem is upon further reading. The Delta log has been cached also and is now stale. You need to run spark.catalog.clearCache and then reads are OK.

Statistical Covariance Part 2

I quoted in a previous post somebody who said that covariance matrices always have a determinant of 0. This isn't strictly true. What can I say? I'm not a very good mathematician.

The argument was that the means are always substracted from each row and with simple algebra you could demonstrate that the determinant is 0. But this ignores the fact that covariance is the expected value of two rows having their expected values subtracted. 

Say we have two different distributions from which we draw combinations. However, for reasons peculiar to the use case, certain combinations are not allowed. When we tabulate the probabilities for this state space, we'll have zeros in some cells. Even though the inner product of the corresponding two probability vectors might not be zero, the expected probability of the two together is.

Note another caveat. Bearing in mind that correlation is just covariance divided by the root of the product of both variances, "the correlation between A and B is only a measure of the strength of the linear relationship between A and B. Two random variables can be perfectly related, to the point where one is a deterministic function of the other, but still have zero correlation if that function is non-linear." [Daniel Shiebler's blog]

Positive Definite

A matrix is positive definite if x^T M x > 0 (an equivalent definition is "a symmetric matrix whose eigenvalues are all positive real numbers" - Coding the Matrix, Klein, definition 12.6.1) 

Note that all matrices that can be expressed as A^TA are positive semi definite (see this SO answer). The proof is simple: substitute A^TA for M above. 

x^T M x = x^T A^TA x = (x A)^T(A x)  

and any non-zero vector or real number multiplied by itself is positive. Since our covariance matrix can be expressed as A^TA it too is at least positive semidefinite (x^T M x ≥ 0). But we know it must also be positive definite as you can't invert M x = 0 for x ≠ 0.

Why this last statement is true can be explained at this elegant SO answer. Basically, if M x = 0 for x ≠ 0 then each row of M must be linarly dependent on each other for the equation 

M _i,j x_j = 0 ∀ i 

to hold. That is, if you give me (n-1) values in a row, I can tell you the value of the last one. Here's a simple Python/Numpy example where the last value is a row is just the sum of the first two. It's easy to see that when multipled with the vector [1, 1, -1] this matrix would be in the null space:

>>> M = np.asmatrix([[1, 2, 3], [4, 5, 9], [6, 7, 13]])
>>> np.linalg.det(M)
0.0

But if a matrix is linearly dependent, it's determinant must be 0. And if a matrix's determinant is 0, it cannot be inverted. QED.

What if my Matrix is Singular?

We add a tiny amount to make it non-singular. This is called conditioning. It sounds like a hack but it does have a basis in maths. In a frequentist interpretation, this is ridge regression. In a Bayesian interpretation, it's the prior.

Briefly, the argument goes that for ridge regression, we penalize large model parameters. So, instead of minimizing our error (θ X - y_actual) we minimize our error plus the penalty:

y_estimate = (θ X - y_actual)² + λ θ^T θ

by differentiating with respect to θ. Solve this equation and you'll see a [X^T X + λ I]^-1.

The argument for the Bayesian prior briefly goes like this: if we take a frequentist view and assume that the error in our data is Gaussian and plug y_estimate into it, we'll see our familiar equation for a Gaussian multiplied by e^λθTθ. Since the Bayesian posterior,  p(θ|Data,Model) must equal the frequentist probability, e^λθTθ is the only term that maps to p(θ|Model) simple because it's the only one with θ in it. Therefore, our conditioning manifests itself in the prior.

Full derivations appear in Jake VanderPlas' wonderful blog.

Scala Proofs - Part 1

According to the Scaladocs,

An instance of A <:< B witnesses that A is a subtype of B.
Requiring an implicit argument of the type A <:< B encodes the generalized constraint A <: B.

So, for instance:

  class MySuper

  class MySub extends MySuper 

  def demandsProof[A](implicit a: A <:< MySuper): Unit = ???

demands proof that any A is a subclass of MySuper. We can ensure that this is satisfied without calling it with an argument:

  demandsProof[MySub] // OK

Rob Norris @tpolecat Apr 02 22:26
A <:< B extends A => B and is available implicitly if A <: B
A =:= B extends A => B and is available implicitly if A is the same type as B

This operator comes in handy when I want to enforce at compile time a rule that says one can only invert square matrices [Wikipedia]. My code looks like this:

@typeclass trait Invert[M[_ <: Length, _ <: Length]] {

  @op("inv") def invert[A <: Length, B <: Length](x: M[A, B])(implicit a: A =:= B): M[A, A]

}

(Note the use of Simulacrum to automatically generate type classes.)

Adam Rosien describes the difference between A <:< B and B >: A at InnerProduct:

There is an analogous implicitly passed value for subtype bounds, but the subtype bound is not syntactic sugar for it. (I thought it was.) That is, the type signature

def widen[A, B >: A](fa: F[A]): F[B] 

is equivalent to the type signature

def widen[A, B](fa: F[A])(implicit ev: A <:< B): F[B] 

but the former is not converted to the latter by the compiler, as is the case for context bounds. [my emphasis]

We can prove this by decompiling the Java bytecode with javap.  The first gives:

  public <A, B> F widen(F);

and the second, context-bound example:

  public <A, B> F widen(F, scala.Predef$$less$colon$less<A, B>);

This highlights a useful equivalence:

Containers of a subtype can be transformed into containers of its supertype, if you can map over the container. The usual defintion of covariance emphasizes subtypes, but the ability to map is a more general, and useful, definition.

Anyway, why would you use one over the other? See the source code for Scala's Option:

sealed trait Option[+A] {
  // def flatten[B, A <: Option[B]]: Option[B] = ...
  // won't work, since the A in flatten shadows the class-scoped A.

  def flatten[B](implicit ev: A <:< Option[B]): Option[B]
    = if(isEmpty) None else ev(get)
  // Because (A <:< Option[B]) <: (A => Option[B]), ev can be called to turn the
  // A from get into an Option[B], and because ev is implicit, that call can be
  // left out and inserted automatically.
}

Odd Characters

The Problem

A data scientist is struggling to get his regex to work in Spark on Databricks. He's wanting to check the data for British postcodes (a.k.a. zipcodes). His regex looks like:

[A-Z]{1,2}[0-9][A-Z0-9]?\\s?[0-9]  

Everybody agrees that any string in the data that matches this regex can reasonably be considered a British postcode. However, he sees in his notebook many instances that look valid but were marked as invalid.

The Investigation

Despite spending most of the last 25 years using the JVM, I've only recently come to appreciate character encodings. It's something that seems annoying and mostly irrelevant. However, last month I wasted two days because the British pound sterling sign (£) can be represented as 0xa3 in what's commonly known as Extended ASCII as well as U+00A3 in Unicode. That means a difference of one byte - which can be the difference between being granted access if it's a password and being locked out if it's not. In this case, the extended ASCII version came from somebody typing on the Linux CLI and the Unicode came from a Java String derived from a properties file. 

Back to our regex. A Scala function seemed to behave differently when it was part of a UDF running on the Spark executor than when it was called locally in the Spark driver. To demonstrate this, the data scientist ran display(hisDataFrame) on the driver and copy-and-pasted the string that should be a British postcode but was not showing up as one into the regex code directly. And it matched! 

What was going on? Cue some false leads and dead ends concerning character encoding (eg, "Unicode grows... char is legacy ... use codePointAt" [SO]; it seems Java hasn't always played fairly with Unicode [SO] even though the code here worked fine on my JDK 1.8). 

But it was this answer on StackOverflow with which I converted the String to hexadecimal. It takes advantage of BigInteger being big-endian when contructed with byte arrays. Converted to Scala it looks like:

def toHex(x: String): String = String.format("0x14x", new java.math.BigInteger(1, x.getBytes))

Now it became clear that there were rogue spaces in the text. So, why had the data scientist's copy-and-paste not worked? Well, Databrick's display function seems to convert the contents of the Dataframe into HTML and multi-spaces in HTML are rendered as just a single space. So, he was pasting the wrong data into his function.

The solution

The solution was to simply add the expectation of mutli-spaces to the regex. But it raised a bigger issue. Although this Databricks display function is great for rendering data in a more pleasing manner, it opens you up to issues like this. Best to stick with DataFrame.show() if you're doing exploratory data analysis and use display for the execs. The former is not subject to this HTML rendering issue.