Polaris and Cloud tokens

Polaris rather pleasingly mints cloud tokens that are subscoped to a directory in a bucket or blob container for AWS and GCP. That is, even if the token has been hijacked, the blast radius is limited by:

• the token only allowing access to a single directory and its subfolders, not the whole bucket
• the token is no good after X minutes (where the default valu of X is 60)

There's currently an outstanding ticket to give subscoping to Azure.

The code for vending for the different clouds belongs in the implementations of PolarisStorageIntegration.getSubscopedCreds and this is where the tokens are created. You could put breakpoints in the breakpoints of:

com.google.auth.oauth2.AccessToken

com.azure.core.credential.AccessToken

software.amazon.awssdk.auth.credentials.AwsSessionCredentials 

and grab the credentials and use them on the command line (that is, entirely outside of Polaris) thus:

# AWS

AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... AWS_SESSION_TOKEN=...  aws s3 ls s3://YOUR_BUCKET/DIRECTORY_FOR_TOKEN

# Azure

az storage blob list  --account-name $STORAGE_ACCOUNT   --container-name $CONTAINER --sas-token $SAS_TOKEN --prefix YOUR_DIRECTORY

# GCP

CLOUDSDK_AUTH_ACCESS_TOKEN=ya29... gcloud storage ls gs://YOUR_BUCKET/DIRECTORY_FOR_TOKEN

But even if you did, the tokens no longer work after 60 minutes and in the case of AWS and GCP, you cannot even view directories for which the token was not defined.

Cloud Topologies and Azure

All clouds follow a similar pattern when it comes to networking. All of them have:

• a VPN (virtual private network) that is an entirely isolated environment
• subnets that are divided into public and private IP spaces by using non-overlapping CIDR blocks
• an internet gateway that allows access to the outside world and performs the NAT
• Route tables that permit traffic flows by typically bringing together subnets and security groups. If subnets are rooms in a building, route tables are the corridors between them and the Network ACLs (NACLs) are the bouncers on the door. Security groups are more identity based and they do a similar job to NACLs albeit at the vNIC level.
• a load balancer that allows incoming traffic from the internet. This differs from the load balancer as it's more the receptionist directing people rather than the security guard on the front door.
• the computers/VMs on which Kubernetes runs, called nodes.

The terminology may slightly differ but this is true for all the main cloud providers

Azure

Now, bearing in mind that this is the general concept for all cloud/K8s clusters, the recipe in Azure is:

1. Create the Resource Group
2. Create an identity for this group.
3. Create a network for this group and identity.

Now, when it comes to the network, the steps are:

1. Create the Virtual Network compatible with the resource group
2. Create the subnet for the Virtual Network
3. Create a Network Security Group for the resource group. This defines the inbound and outbound rules. Funnily enough, the rules in the NSG are of higher priority the lower the number.
4. We associate the subnet with the Network Security Group
5. We assign a role to the subnet.

Now we can set up the Kubernetes cluster if we:

1. Assign the subnet to a K8s Node Pool
2. Create a network profile

This last one is interesting. It's the "the reference to the network interface card" [Azure for Architects, Packt]. The network car is the actual physical hardware the packets must use but generally they're only referenced in Azure (and GCP) as virtual network interface cards (vNIC). In AWS, they're called ENIs.

Cloud Architecture

This is still something I am mulling over but here are my notes.

Problem statement

We want different organisations to share large quantities of confidential data to be processed.

The prerequisites are:

1. must be secure
2. must be entirely FOSS based
3. must be cross-cloud
4. allows a bring-your-own policy

The choice of Apache Iceberg for the data seems to be straightforward. But the question of infra provisioning is not an easy one with a whole debate going on in Discord. Some love Terraform for being (somewhat) typesafe, others think controlllers are the way to go.

Infra provisioning

As ever, the answer to what route to take is "it depends" but here are some of the esoteric terms defined.

Crossplane is a CNCF-compliant, Golang "backend that enables you to build a control plane that can orchestrate applications and infrastructure no matter where they run". So, you could use Crossplane to provision infra not just in its K8s cluster but in the cloud. ACK (AWS Controllers for K8s) is an AWS specific equivalent of Crossplane that watches its CRDs and provisions accordingly.

In the other corner is the reigning champion, Terraform and it's FOSS fork, OpenTofu (both written in Go). Terraform has a form of type system but it's not enforced until the plan stage and is "loose" as it's not strict but allows type coercion. 

You can use CDKTF (which has common language bindings to create Terraform config files) but there is some doubt about its future.

Another tool to address the issues with raw Terraform (lack of DRY principles, ciruclar dependencies, orchestration etc) is Terragrunt, a thin wrapper around Terraform/OpenTofu written in Go. It allows the output from one stage to be the input to another [Discord]. Hashicorp, the creators of Terraform, have recognised these problems and have released Stacks.

A central way to orchestrate and deploy your Terraform config is the (mostly) Java Terrakube. It also adds an RBAC layer. Because everything runs remotely, it can be scheduled to detect drift, say, in the middle of the night.

Similarly, Atlantis is a Go server that has RBAC, executes the commands, requires approval for pull requests before applying them and generally implements a GitOps workflow.

For self-serve, there is the FOSS, Typescript Kubero. This allows developers to manage their own K8s deployments. Coolify, meanwhile, is a PHP tool that makes managing your own servers very cloud-like. It just needs an SSH connection.

Lessons from a migration

Don't use low/no-code. It really is a false economy. It appeals to managers during sales pitches because it looks so simple. But your team will spend all their time and all your money debugging in a most un-ergonimc manner. And as for testing....

Azure Data Factory is a no-code solution that's great for simple DAGs. However, it is not suitable for anything more entrerprise-y. For instance, if you're copying large amounts of data from one database to another and there is a failure, there is no scope for cleaning up the data. The first you'll know of it (because nobody ever reads the logs) is when the users are complaining that the numbers coming out of their SQL queries are wrong.  

ADF doesn't even allow nested ForEach loops! It stores pipelines as JSON so watch it get confused when the pipeline itself contains JSON!

Don't give people raw SQL access. They can make profound changes and you'll have no logs and little ability to correct it.

Everything needs to have an automated audit log. There's going to be large number of questions like: "why did it do this?" It's never simply a matter of success or fail. There is huge nuance - eg, type conversion may mean what was in the source system is not exactly the same as in the destination system. Is this a pass or a fail?

Processes need orchestrating. One process reads a dataset while another deletes it/writes to it causing nondeterministic errors. You get similar issues when you issue a cancel.

Communication - docs are always changing. Be aware that most knowledge is tribal, not recorded.

Scheduling: everything was based on time. So, it's possible two daily jobs were running at the same time if the first took more than 24 hours. Data migrating from one DB to another within a cloud zone and subscription was at a rate of ~10mb/s. This meant some tables took hours. And it didn't scale. As the project progressed, more tables were to be migrated. But this caused some jobs to take days. So, the weekly transfer was still going on when people came to the office on Monday morning. Consequently, their queries were returning inconsistent results - the silent killer.

The metadata must be constrained. If it lives in a database, it must have referential integrity. This can be overlooked because it's not business data. But if you want to eliminate mysterious errors, it's essential to get this right.

Like water finding its natural level, people will naturally gravitate to the path of least resistance. As a general rule, teams do not follow best practises, naming conventions or industry standards. This is not strictly true but it's so common that you must assume for all teams that their data and code is a Rube Goldberg Machine.

Regular meetings with major stakeholders. These meeting can be brief (about 15 minutes is OK once the cadence is established) but they do need to be frequent (at least twice a week, preferably daily).

Architecting Azure

Nineteen hours into a job migrating data from Synapse to an Azure SQL Server, we see: 

Failure happened on 'Source' side. ErrorCode=SqlOperationFailed,'Type=Microsoft.DataTransfer.Common.Shared.HybridDeliveryException,Message=A database operation failed with the following error: 'A transport-level error has occurred when receiving results from the server. (provider: TCP Provider, error: 0 - The specified network name is no longer available.)',Source=,''Type=System.Data.SqlClient.SqlException,Message=A transport-level error has occurred when receiving results from the server. (provider: TCP Provider, error: 0 - The specified network name is no longer available.),Source=.Net SqlClient Data Provider,SqlErrorNumber=64,Class=20,ErrorCode=-2146232060,State=0,Errors=[{Class=20,Number=64,State=0,Message=A transport-level error has occurred when receiving results from the server. (provider: TCP Provider, error: 0 - The specified network name is no longer available.),},],''Type=System.ComponentModel.Win32Exception,Message=The specified network name is no longer available,Source=,'

Yikes. This is after 226gb and 165 million rows have been written at an average throughput of 3.3MB/s. Three copy activities stopped within three seconds of each other but nothing untoward was found in the AzureDiagnostics and AzureActivity logs. At first I thought the network was suspiciously quiet at the time the copy came to an end but with the Azure logs and this Pandas code here, I found that brief pauses were not that unusual:

Bursty network logs

Other engineers said they see this intermittently. "Welcome to the world of cloud computing where Transient Faults are bound to happen" [SO]. A cloud solution architect at Microsoft writes that throttling may be"done via blocking the connections or denying the new connections to SQL Azure database engine". Or it could be the network. "In Azure, most of the components are running on the internet, and that internet connection can produce transient faults intermittently." [Azure for Architects]

Never assume that a network is reliable, whether it be the cloud or not. I worked in an investment bank where a developer would make a connection to a system and if it was connected, assume the failover system was live (this was a blue/green deployment). At first blush, this was not unreasonable but networks can be tricksy. As it happened, the network must have hiccuped and he was connected to the standby system pumping live data into it. 

The Death of Data Locality?

Data locality is where the computation and the storage are on the same node. This means we don't need to move huge data sets around. But it's a pattern that has fallen out of fashion in recent years.

With a lot of cloud offerings, we lose the data locality that made Hadoop such a great framework on which to run Spark some 10 years ago. The cloud providers counter this with a "just rent more nodes" argument. But if you have full control over your infra, say you're on prem, throwing away data locality is a huge waste.

Just to recap, data locality gives you doubleplusgood efficiency. Not only does the network not take a hit (as it doesn't need to send huge amoungs of data from storage to compute nodes) but we retain OS treats like caching. 

What? The OS has built in caching? Have you ever grepped a large directory and then noticed that executing the same command a second time is orders of magnitude faster than the first time? That's because modern operating systems leave pages in memory unless there is a reason to dispose of them. So, most of the time, there is no point in putting some caching layer on the same machine as where the database lives - a strange anti-pattern I've seen in the wild.

Of course, none of this is not available over the network.

Another advantage of having the data locally is that apps can employ a pattern called "memory mapping". The idea is that as far as the app is concerned, a file is just a location in memory. You read it just like you would a sequence of bytes in RAM. Hadoop takes advantage of this.

Why is memory mapping useful? Well, you don't even need to make kernel calls so there is no context switching and certainly no copying data. Here is an example of how to do this in Java. You can prove to yourself that there are no kernel calls by running:

sudo strace -p $(jstack $(jps | grep MemoryMapMain | awk '{print $1}')  | grep ^\"main | perl -pe s/\].*\//g | perl -pe s/.*\\[//g)

Note there are kernel calls in setting up the memory mapping but after that, there is nothing as we read the entire file.

So, why have many architects largely abandoned data locality? It's generally a matter of economics as the people at MinIO point out here. The idea is that if your data is not homogenous, you might be paying for, say, 16 CPUs on a node that's just being used for storage. An example might be that you have a cluster with 10 years of data but you mainly use that last two years. If the data for the first eight years is living on expensive hardware and rarely accessed, that could be a waste of money.

So, should you use data locality today? The answer, as ever, is "it depends".

Architectural patterns

Some architectural terms (old and new) that I keep bumping into.

Eventual Consistency

"Eventual consistency — also called optimistic replication — is a consistency model used in distributed computing to achieve high availability that informally guarantees that, if no new updates are made to a given data item, ultimately all accesses to that item will return the last updated value.  Eventually-consistent services are often classified as providing BASE semantics (basically-available, soft-state, eventual consistency), in contrast to traditional ACID ... Another great model that can be directly implemented in the application layer is strong eventual consistency (SEC), which can be achieved via conflict-free replicated data types (CRDT), giving us the missing safety property of eventual consistency.  

"Event-driven applications usually favor eventual consistency, for the most part. However, we could also opt-in for strong consistency in particular system areas. Thus, it is fair to say we can combine both consistency models depending on our use case." - Functional Event-Driven Architecture, Volpe

The impacts of consistency on Microservices

Microservices should ideally be totally independent. For example, in a highway  management system, the weather service is totally orthoganol to the roadworks service even though both have an impact on congestion. However, microservices in the real world often have soft dependencies. As a result, "in a microservices world, we don’t have the luxury of relying on a single strongly consistent database. In that world, inconsistency is a given." [James Roper]

Hugo Oliviera Rocha outlines some antipatterns here. The first is "events as simple notifications. The source system publishes an event notifying the consumers that something changed in its domain. Then the consumers will request additional information to the source system... The main issue and the main reason why this option should be seldom used is when you apply it to a larger scale.

"[I]nstead of requesting the source system for additional information, it is possible to save the data internally as a materialized read model... The main issue isn’t the disk space, it is the initialization, maintenance, and keeping that data accurate."

He says event sourcing is just a band aid and suggests using fat (ie, denormalised) messages. The downside it they can be chunky.

CRDT

"To implement eventually consistent counting correctly, you need to make use of structures called conflict-free replicated data types (commonly referred to as CRDTs). There are a number of CRDTs for a variety of values and operations: sets that support only addition, sets that support addition and removal, numbers that support increments, numbers that support increments and decrements, and so forth." - Big Data, Nathan Marz

To a functional programmer, this looks a lot like semigroups and reducing.

Data Mesh

"Unlike traditional monolithic data infrastructures that handle the consumption, storage, transformation, and output of data in one central data lake, a data mesh supports distributed, domain-specific data consumers and views “data-as-a-product,” with each domain handling their own data pipelines. The tissue connecting these domains and their associated data assets is a universal interoperability layer that applies the same syntax and data standards." [TowardsDataScience]

"Data Mesh is a journey so you cannot implement Data Mesh per-se, you need to adopt the principles and start to make incremental changes." Adidas's journey [Medium]. Of the seven points given, two (decentralization and self-service) are the antithesis of ontologies.

Batch Views
"The batch views are like denormalized tables in that one piece of data from the master dataset may get indexed into many batch views. The key difference is that the batch views are defined as functions on the master dataset. Accordingly, there is no need to update a batch view because it will be continually rebuilt from the master dataset. This has the additional benefit that the batch views and master dataset will never be out of sync."  Big Data, Nathan Marz

Saga Pattern
"The Saga Pattern is as microservices architectural pattern to implement a transaction that spans multiple services. A saga is a sequence of local transactions. Each service in a saga performs its own transaction and publishes an event. The other services listen to that event and perform the next local transaction" [DZone]

Example in Cats here.

Type 1 and 2 data evolution

Slowly changing dimensions [Wikipedia] is a "concept that was introduced by in  Kimball and Ross in The Data Warehouse Toolkit."  A strategy could be that the data source "tracks historical data by creating multiple records. This is called a type 2 dimension." [The Enterprise Big Data Lake - Gorelik].  

Type 1 is overwritting a row's data as opposed to type that adds a new row.

Data Marts

Definitions for data marts tend to be a bit wooly but the best I heard was from a colleague who defined it as "data structured for use cases and particularly queries."

Data Marts tend to use type 2 dimensions (see above). 

Hexagon Architecture
Hexagon a.k.a Onion a.k.a Ports and Adapters "give us patterns on how to separate our domain from the ugliness of implementation." [Scala Pet Store on GitHub] This is an old pattern, as anybody who has written microservices will know, but the name was new to me.  The idea is that there are many faces the app shows the outside world for means of communication but the kernel inside "is blissfully ignorant of the nature of the input device." [Alistair Cockburn] This faciliates testing and reduces cognitive overhead that comes from having business logic scattered over many tiers and codebases.

Microservices

This is a huge area but here are some miscellaneous notes.

Before you jump on board with the Java based Lagom, it's worth noting that Martin Fowler wrote "Don't start with microservices – monoliths are your friend". This provoked a whole debate here. It's all worth reading but the comment that stuck out for me was:

"Former Netflix engineer and manager here. My advice:

Start a greenfield project using what you know ... Microservices is more often an organization hack than a scaling hack. Refactor to separate microservices when either: 1) the team is growing and needs to split into multiple teams, or 2) high traffic forces you to scale horizontally. #1 is more likely to happen first. At 35-50 people a common limiting factor is coordination between engineers. A set of teams with each team developing 1 or more services is a great way to keep all teams unblocked because each team can deploy separately. You can also partition the business complexity into those separate teams to further reduce the coordination burden."

A fine example of Conway's Law.

Builds in large organisations

Interestingly, Facebook report Git not being scalable. Meanwhile, Google uses Bazel which is supposed to be polyglot and very scalable.

Strangler Pattern

This is one of those obvious patterns that I never knew had a name.

"The Strangler pattern is one in which an “old” system is put behind an intermediary facade. Then, over time external replacement services for the old system are added behind the facade... Behind the scenes, services within the old system are refactored into a new set of services." [RedHat]

Downsides can be the maintenance effort.

Medallion architecture

This [DataBricks] divides data sets into bronze (raw), silver (cleaned) and gold (application-ready).

Journeys in Data Engineering

I'm currently helping a huge health provider with its data issues. Here are some things I've learned:

Don't bikeshed. There's no point wondering whether you've captured the correct ethnic description ("White? Irish White? Eastern European White?") when there are bigger fish to fry. For instance, hospitals were putting the patients' personal information in the primary key field. This sent the data officer apoplectic until it was cleansed from the files before being sent downstream. But as a result, the downstream keys were not unique. The data scientists consuming this data aggregated it and came to conclusions unwittingly based on one individual having visited the hospital over three million times.

Don't proactively go looking for data quality issues. They'll come looking for you. Continue to build your models but be extremely circumspect. This is a more efficient process than spending time wondering if the data looks good.

Just because the data looks good, it doesn't mean it's usable. How often is the data updated? Is it immutable or is it frequently adjusted? In short, there's an orthogonal axis in data space separate to data quality and it's a function of time. Perfect data that becomes old is (probably) no longer perfect. Perfect data that becomes incomplete is (probably) no longer perfect.

Give a thought to integration tests. Platforms like Palantir are pretty under-developed in this area (answer to my requests for an integration test platform: "we're working on it"). So, you may need to write bespoke code that just kicks the tires of your data every time you do a refresh. 

Remember that documentation is the code. I've had a nice experience using ScalaTest with its Given, When, Thens. It ensured that when running integration tests, the code generated the documentation so the two would never fall out of synch. This is, of course, much harder (impossible?) when running in a walled garden like Palantir.

Stick with a strongly typed language. This can be hard since pretty much all data scientists use Python (and therefore PySpark) but there is nothing worse than waiting several minutes for your code to run only to found out that a typo trips you up. In a language that is at least compiled, such problems would not occur. I'd go so far to say that Python is simply not the correct tool for distributed computing.

Python tools like Airflow are made much easier by using Docker since Python library version management is a pain in the arse. Far better for each application to have its own Docker container.

Never forget that your schema is your data contract as much as an API is your coding contract. Change it, and people downstream may squeal.

Persisting intermediate data sets hugely help debugging. 

Finally, don't use null to indicate something for which you know the meaning. If, for instance, when a record is in a certain state, it's better to say value X equals a flag to indicate that. If you use null, somebody reading that data doesn't know if the record is in this state or if the data is just missing.

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.

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.

Mathematical Proofs for Distributed Systems

Software is becoming more about proofs. With well written FP code, you can prove at compile time that your software cannot enter certain pathological states.

Well, the world of architecture is moving in that direction, too.


Introducing TLA+

TLA+ is an attempt to help architects and developers to write bullet-proof systems. It's particularly useful in the domain of distributed computing.

"Distributed algorithms and protocols are hard to get right, especially, when they have to tolerate faults. One of the reasons is that distributed algorithms vary in their assumptions about thedistributed system: the communication medium, system synchrony, possible faults, etc. ... [TLA] offers a rich syntaxfor sets, functions, tuples, records, and sequences on top of first-order logic"

TLA+ Model Checking Made Symbolic

It's already being used in architecting Kafka.


Propositional Logic

In propositional logic, "there are only two values, TRUE and FALSE. To learn how to compute with these values, all you need to know are the ... definitions of the five Boolean operators"[1]: ∧, ∨, ¬, ≡ and ⇒.

The first four are AND, OR, NOT and EQUALS, so nothing special there. The last one (the implies operator) is tricksy. Its truth table looks like this:

F	G	F⇒G
TRUE	TRUE	TRUE
FALSE	TRUE	TRUE
FALSE	FALSE	TRUE
TRUE	FALSE	FALSE

To help understand it, let's look at this example formula:

(n > 3) ⇒ (n > 1)

The first three rows of the truth table simply describe n>3,  1<n≤3 and n≤1.

The last row is when n>3 but not n>1 which is clearly wrong, hence FALSE.

Now, say you were asked to prove that:

(F⇒G) ≡ (¬F∨G)

is a tautology. You could write out all the truth tables which would be laborious. "However, computers are better at doing this sort of calculation."[1]


Very nice, but what's the application?

You can find a very nice model of some Kafka functionality here on GitHub. You can open this in the IDE called TLA+ Toolbox. It looks like this:

The language is pretty straight forward. For instance, pre-pending formulas with operators like /\ is syntactic sugar for removing wide parenthesis and EXTENDS seems to be like import in Java and Scala. You can find some blogs to learn more about it here [LearnTLA], here [Anton Sookocheff's blog] and here [Jack Van Lightly's blog].

The Toolbox can turn this into a more readable PDF (more readable for mathematicians anyway):

You can then "run" your model by giving it initial parameters. The Toolbox will then explore its state space. Interestingly, Hillel Wayne in this video descibes the path through the states as a description of the behavious of a system.

Another video can be found here and the transcript to Lamport's own presentation can be found here.

Conclusion

TLA+ looks like it could turn the IT architecture industry into something more than "it feels kinda right" that we see at the moment. One only hopes that its adoption in projects like Kafka will encourage other people to learn it. Me - I've only taken the first few steps. 

[1] Specifying Systems, Dr Leslie Lamport

Architecting the Cloud

We're taking the plunge and moving a SOC to Google's cloud. This is the proposed architecture. With von Moltke's warning of "no plan survives first contact with the enemy" in mind, I write this for posterity and we shall see how well it stands up to reality.

Architecture

Ingestion

Tools: Apache Kafka.

Why? Because Kafka (unlike Google's PubSub) is not tied to a particular cloud provider. What's more, it's the de facto standard for high volume streams of data and finding developers who know it is not hard.

Storage

Tools: Google's BigQuery and DataProc.

Why? Google will worry about the long term storage of the data in BigQuery and we will pull subsets of it out into DataProc (which is basically Apache Spark, YARN and Hadoop) for analysis. This is Google's recommend way of doing it although one must bear in mind that they are trying to sell something. This basically makes BigQuery the golden source of data and DataProc the data lake.

Visualisation

Tools: Google's DataLab (aka, Jupyter) and some DataProc components (specifically, Apache Zeppelin)

Why? Jupyter and Zeppelin are open source. Zeppelin integrates very nicely with Spark, er, I mean Google DataProc; and Jupyter integrates easily with BigQuery.


Potential Use Cases

The Senior SOC Analyst - Investigation

A sufficiently sophisticated analyst should have no problem writing Jupyter notebooks that access BigQuery directly using plain SQL. They may need some familiarity with Python if they then wish to present this data.

The Junior SOC Analyst - Investigation

Google does not appear to have a managed Elastic component. However, plain old Linux instances can be spun up and Elastic installed on them manually. This Elastic cluster can then batch load data that was extracted from BigQuery. Alternatively, it could stream from the Kafka topic used in the ingestion layer. Either way, the analyst then can run free-text searches over the event data rather than using the more complicated SQL and Python languages.

The Data Scientist - Modelling

Given a requirement to build a deliverable, a data scientist can spin up an appropriately sized DataProc cluster and transfer data from BigQuery. BQ will be his data lake.

"Simply bringing data from various parts of a large organization together in one place is valuable because it enables joins across datasets that were previously disparate ... collecting data in its raw form, and worrying about schema design later, allows the data collection to be speeded up (a concept sometimes known as a data lake or enterprise data hub... Simply dumping data in its raw form ... has been dubbed the sushi principles: raw data is better". [1]

So, imagine that the data scientist is looking at anomalous employee behaviour. He can pull in large amounts of data from BigQuery in an arbitrary format. He can also pull in the employee database and marry the two together in DataProc. By itself, the BigQuery data won't necessarily tell you that an SSH connection is suspicious. However, coupled with employee data, it would be highly suspicious if that SSH connection came from the machine of a 52-year old HR lady rather than a 25-year old sysadmin.

The Data Scientist - Producing Artefacts

Google's Spark/Hadoop/YARN offering (DataProc) only holds ephemeral data. When the cluster is no longer needed, it can be closed down and the data will disappear into the ether but not before the Data Scientist has finished building his model. This model will be distilled from the petabytes in BigQuery and the terabytes in DataProc and can be brought out of the cloud inexpensively in the form of a moderately sized (megabytes?) artefact.

The Developer - Writing Parsers

Feeds into the ingestion system can be of disparate formats (CEF, Rsyslog etc). Therefore, developers must be able to write new parsers and deploy them. It makes sense to run these parsers on the ingestion layer rather than on all the agents to avoid deployment faff.

The Developer - Analysing Streams

Since analysts want to be warned of high-risk events as soon as possible, there will be a need for the developers to write stream monitoring software. This may listen to the same Kafka topic used in ingestion. Perhaps it can use Google's PubSub to send the warnings if it lives in the cloud.

The Developer - Using ML Models in Production

It is to be hoped that artefacts generated by the data scientists can be deployed in production. These artefacts should be written in such a way that they do not need to access the data in GCP. For instance, a data scientist might train a neural net in GCP to spot anomalous behaviour then, when happy with it, give it to the developer (preferably as a JAR) to use in their stream monitory software.

Schema

The ingestion layer can do the following to the data before feeding it into BigQuery:

1. Do no parsing and put it into BigQuery raw.
2. Partially parse it.
3. Parse it completely.

Each has pros and cons.

No Parsing

This is the simplest path initially. It means all data lives in the same table irrespective of which stream it came from. But it also means that BigQuery will have no schema to speak of, the classic pattern of "schema-on-read [where] the structure of the data is implicit, and only interpreted when the data is read" [1].

This will make querying and data extraction extremely difficult as the table will look a little something like this:

IngestTimestamp	Value
Thu  6 Jun 16:44:01.123 BST 2019	{ "additionalEventData": { "SSEApplied": "SSE_S3", "x-amz-id-2": "u8yUG+gwxxc3ovA49cFbD9XFLyHRBJAkwnHPeFZVg4AnfGKp30LXiQ2oAY3BUX5Lwflkijiaq9M=" }, "awsRegion": "eu-west-1", "eventID": "e4490a9a-1be4-4585-98ee-9d0a4e59fcc3", "eventName": "PutObject", "eventSource": "s3.amazonaws.com", "eventTime": "2018-12-05T14:47:26Z", "eventType": "AwsApiCall", "eventVersion": "1.05", "readOnly": false, "recipientAccountId": "479626555249", "requestID": "770130FED723D6EA", "requestParameters": { "bucketName": "vsecacloudtraillogs", "key": "AWSLogs/479626555249/CloudTrail/eu-west-1/2018/12/05/479626555249_CloudTrail_eu-west-1_20181205T1450Z_apgVO8lFdNcrVfHn.json.gz", "x-amz-acl": "bucket-owner-full-control", "x-amz-server-side-encryption": "AES256" }, "resources": [ { "ARN": "arn:aws:s3:::vsecacloudtraillogs/AWSLogs/479626555249/CloudTrail/eu-west-1/2018/12/05/479626555249_CloudTrail_eu-west-1_20181205T1450Z_apgVO8lFdNcrVfHn.json.gz", "type": "AWS::S3::Object" }, { "ARN": "arn:aws:s3:::vsecacloudtraillogs", "accountId": "479626555249", "type": "AWS::S3::Bucket" } ], "responseElements": { "x-amz-server-side-encryption": "AES256" }, "sharedEventID": "bfa6ef38-d494-4974-8d76-1c07b305ae90", "sourceIPAddress": "192.168.0.1", "userAgent": "cloudtrail.amazonaws.com", "userIdentity": { "invokedBy": "cloudtrail.amazonaws.com", "type": "AWSService" } }
Thu  6 Jun 16:44:01.124 BST 2019	{ "region": "us-east-1", "detail": { "type": "UnauthorizedAccess:EC2/SSHBruteForce", "resource": { "resourceType": "Instance", "instanceDetails": { "instanceId": "i-99999999", "instanceType": "m3.xlarge", "launchTime": "2016-08-02T02:05:06Z", "platform": null, "productCodes": [ { "productCodeId": "GeneratedFindingProductCodeId", "productCodeType": "GeneratedFindingProductCodeType" } ], "iamInstanceProfile": { "arn": "GeneratedFindingInstanceProfileArn", "id": "GeneratedFindingInstanceProfileId" }, "networkInterfaces": [ { "ipv6Addresses": [], "networkInterfaceId": "eni-bfcffe88", "privateDnsName": "GeneratedFindingPrivateDnsName", "privateIpAddress": "10.0.0.1", "privateIpAddresses": [ { "privateDnsName": "GeneratedFindingPrivateName", "privateIpAddress": "10.0.0.1" } ], "subnetId": "Replace with valid SubnetID", "vpcId": "GeneratedFindingVPCId", "securityGroups": [ { "groupName": "GeneratedFindingSecurityGroupName", "groupId": "GeneratedFindingSecurityId" } ], "publicDnsName": "GeneratedFindingPublicDNSName", "publicIp": "127.0.0.1" } ], "tags": [ { "key": "GeneratedFindingInstaceTag1", "value": "GeneratedFindingInstaceValue1" }, { "key": "GeneratedFindingInstaceTag2", "value": "GeneratedFindingInstaceTagValue2" }, { "key": "GeneratedFindingInstaceTag3", "value": "GeneratedFindingInstaceTagValue3" }, { "key": "GeneratedFindingInstaceTag4", "value": "GeneratedFindingInstaceTagValue4" }, { "key": "GeneratedFindingInstaceTag5", "value": "GeneratedFindingInstaceTagValue5" }, { "key": "GeneratedFindingInstaceTag6", "value": "GeneratedFindingInstaceTagValue6" }, { "key": "GeneratedFindingInstaceTag7", "value": "GeneratedFindingInstaceTagValue7" }, { "key": "GeneratedFindingInstaceTag8", "value": "GeneratedFindingInstaceTagValue8" }, { "key": "GeneratedFindingInstaceTag9", "value": "GeneratedFindingInstaceTagValue9" } ], "instanceState": "running", "availabilityZone": "GeneratedFindingInstaceAvailabilityZone", "imageId": "ami-99999999", "imageDescription": "GeneratedFindingInstaceImageDescription" } }, "service": { "serviceName": "guardduty", "action": { "actionType": "NETWORK_CONNECTION", "networkConnectionAction": { "connectionDirection": "INBOUND", "remoteIpDetails": { "ipAddressV4": "127.0.0.1", "organization": { "asn": "-1", "asnOrg": "GeneratedFindingASNOrg", "isp": "GeneratedFindingISP", "org": "GeneratedFindingORG" }, "country": { "countryName": "GeneratedFindingCountryName" }, "city": { "cityName": "GeneratedFindingCityName" }, "geoLocation": { "lat": 0.0, "lon": 0.0 } }, "remotePortDetails": { "port": 32794, "portName": "Unknown" }, "localPortDetails": { "port": 22, "portName": "SSH" }, "protocol": "TCP", "blocked": false } }, "resourceRole": "TARGET", "additionalInfo": { "sample": true }, "eventFirstSeen": "2018-05-11T14:56:39.976Z", "eventLastSeen": "2018-05-11T14:56:39.976Z", "archived": false, "count": 1 }, "severity": 2, "createdAt": "2019-06-06T16:50:11.441Z", "updatedAt": "2018-05-11T14:56:39.976Z", "title": "127.0.0.1 is performing SSH brute force attacks against i-99999999. ", "description": "127.0.0.1 is performing SSH brute force attacks against i-99999999. Brute force attacks are used to gain unauthorized access to your instance by guessing the SSH password." } }

That is, all the data in one, big varchar if we were to use RDBMS parlance. Note how the table does not care whether the event is GuardDuty or a CloudTrail. Also note that although in this example it happens that GuardDuty and CloudTrail are JSON and a schema can be inferred, not all streams are going to be in the JSON format. In any case "if your BigQuery write operation creates a new table, you must provide schema information" (from the docs).

So, how would you write a query to, say, pull out a record with a given invokedBy field? And how would you limit that query to just a subset of the data (to reduce processing costs)?

It appears that BQ allows UDFs (user defined functions) but they're written in JavaScript. If you're going to write a parser, would you rather write and test it locally in a JVM language or in JS running on an opaque technology stack?

Partial Parsing

The idea here is to "lightly" parse the data. The fields that are common over all streams are event time, machine name and stream name - that is all (this appears to be what the Apache Metron team have gone with. "We are currently working on expanding the message standardization beyond these fields, but this feature is not yet available" - from here).

These parsed fields can be pulled out into their own columns but the payload must remain a string. So, like the solution above (No Parsing), analysts will still find it hard to grok the data if they access it directly via BQ.

However, things may change upon pulling the data from BQ to, say DataProc. In this case, only what is needed (given time range, machine and/or stream type) will be pulled out and bespoke parsers can be written that pull only the data needed for the query. This will require far less development work than the Complete Parsing solution below but will require an increased level of sophistication from the analysts.

Complete Parsing

This requires writing (or using open source) parsers for a particular feed and then mapping it to a rich schema in BigQuery. This will require a lot of up-front development work but it results in full, rich SQL queries than can be run by the analysts.

This solution would probably lead to a different table per event stream type as they all represent different data. If you were feeling particularly ambitious, you could try to unify all this data into one table. However, the history of attempting to unify all types of security events into one schema has been a sorry tale. Rafael Marty, who has spent his career trying to do exactly this says:

"For over a decade, we have been trying to standardize log formats. And we are still struggling. I initially wrote the Common Event Format (CEF) at ArcSight. Then I went to Mitre and tried to get the common event expression (CEE) work off the ground to define a vendor neutral standard. Unfortunately, getting agreement between Microsoft, RedHat, Cisco, and all the log management vendors wasn’t easy and we lost the air force funding for the project. In the meantime I went to work for Splunk and started the common information model (CIM). Then came Apache Spot, which has defined yet another standard (yes, I had my fingers in that one too). So the reality is, we have 4 pseudo standards, and none is really what I want. I just redid some major parts over here at Sophos (I hope I can release that at some point)."

So, only the brave, foolhardy or well-funded need apply.

Schema evolution

Note that it seems that the scope for schema evolution is limited in BigQuery which "natively supports [only] the following schema modifications: Adding columns to a schema definition; Relaxing a column's mode from REQUIRED to NULLABLE" (from the docs).

This means that having a rich schema (as in Complete Parsing) that is easy to interrogate is unfortunately very brittle. You must have your schema pretty much nailed down on your first attempt or else you may very well be forced to introduce hacks further down the line as you shoe-horn changes into it.

Which approach is best?

"Enforcement of schemas in database is a contentious topic, and in general there's no right or wrong answer." [1]

Conclusion

I'm firmly in the camp that says the architecture document should come at the end of an MVP not the beginning. It must adapt to the hidden realities that become apparent along the way. I'll revisit this post in the future to see what actually happened in our case.

We started with a military quote and we'll end with one, this time from Eisenhower: "In preparing for battle I have always found that plans are useless, but planning is indispensable."

[1] Designing Data Intensive Applications, Klepperman

HBase, Cassandra and CAP

Which is best - Cassandra or HBase? Well, it depends what you're trying to do.

The oft-quoted comparisons generally involve Eric Brewer's CAP-theorem (the formal proof by Gilbert and Lynch can be found here and an updated review here) and you can see pictures like this dotted around the web:

From "Cassandra: the Definitive Guide" quoted here

(where the RDMBS offerings imply Two-Phased Commit).

Most developers have some awareness of CAP but there are a lot of misconceptions - typically that you can have at most two elements of Consistency, Availability or Partition tolerance - but not all three. "The 2 of 3 formulation was always misleading because it tended to oversimplify the tensions among properties" (from an article by Brewer himself found here).

Allowing at least one node to update state will cause the nodes to become inconsistent, thus forfeiting C. Likewise, if the choice is to preserve consistency, one side of the partition must act as if it is unavailable, thus forfeiting A. Only when nodes communicate is it possible to preserve both consistency and availability, thereby forfeiting P. The general belief is that for wide-area systems, designers cannot forfeit P and therefore have a difficult choice between C and A.

Here are some common misconceptions corrected.

• The C in CAP "has got nothing to do with the C in ACID, even though that C also stands for 'consistency'" [1]. "Gilbert and Lynch use the word “atomic” instead of consistent in their proof, which makes more sense" [2]. The C in ACID has several similar but not mutually exclusive meanings which include not violating any database constraints, linearizability (see below), ensuring the data is consistent (eg, domain-specific rules such as taking X out of one bank account and into another means the total money overall does not change).
• "The CAP model says nothing about transactions that touch multiple objects. They are simply out of scope for the theorem" [2].

In "Please stop calling databases CP or AP", Martin Kleppmann says:

"But if you’re using some other notion of consistency or availability, you can’t expect the CAP theorem to still apply. Of course, that doesn’t mean you can suddenly do impossible things, just by redefining some words! It just means that you can’t turn to the CAP theorem for guidance, and you cannot use the CAP theorem to justify your point of view.

"If the CAP theorem doesn’t apply, that means you have to think through the trade-offs yourself. You can reason about consistency and availability using your own definitions of those words, and you’re welcome to prove your own theorem. But please don’t call it CAP theorem, because that name is already taken."

In the CAP theorem:

• Consistent means Linearizability,  think of visibility in terms of variables in the JVM. "If operation B started after operation A successfully completed, then operation B must see the the system in the same state as it was on completion of operation A, or a newer state." [1] A master that asynchronously replicates to slaves is an example of a non-linearizable system. Note: some systems (eg a DB that uses MVCC) are intentionally non-linearizable.
  
  
• Availability means “every request received by a non-failing [database] node in the system must result in a [non-error] response” [from the proof]. Note, the response can take an arbitrary amount of time which might violate some people's notion of availability...
  
  
• "Partition Tolerance (terribly mis-named) basically means that you’re communicating over an asynchronous network that may delay or drop messages." Which is basically the internet. "
  
  "In practical terms, a distributed system cannot be made immune to network failures... If the system chooses to favor Availability (such systems are designated AP systems), then it will continue to service requests, even though the data could be in an inconsistent state... If the system chooses to favor Consistency (known as a CP system), then it will choose to stop serving requests (become unavailable) if the consistency of data cannot be guaranteed. For a CP system, this is achieved by requiring a certain number of nodes to confirm that a data update has been made before acknowledging the update." [Hazelcast]

Using these definitions, Cassandra is not AP for quorum read/writes. If there is a partition split, the nodes on the wrong side of the split cannot reach a consensus and therefore are not available. (Interestingly, Brewer notes "as some researchers correctly point out, exactly what it means to forfeit P is unclear.")

So, although Cassandra and HBase both have their pros and cons, mentioning the CAP theory is orthogonal.

(Aside: although Elastic Search is a search engine not a database, there is an interesting chat about how CAP applies to it here.)

A solution?

There's a cheekily entitled blog post (it's titled "How to beat the CAP theorem" but in it he says "You can't avoid the CAP theorem, but you can isolate its complexity and prevent it from sabotaging your ability to reason about your systems"). This isolation is via immutability.

In this post, author Nathan Marz talks of DBs "choosing availability over consistency":

The best consistency guarantee these systems can provide is eventual consistency. If you use an eventually consistent database, then sometimes you'll read a different result than you just wrote. Sometimes multiple readers reading the same key at the same time will get different results... It is up to you to repair the value once you detect that the values have diverged. This requires tracing back the history using vector clocks and merging the updates together (called read repair)

Marz' proposal to use immutable data leads us to this conclusion:

If you choose consistency over availability, then not much changes from before. Sometimes you won't be able to read or write data because you traded off availability... Things get much more interesting when you choose availability over consistency. In this case, the system is eventually consistent without any of the complexities of eventual consistency. Since the system is highly available, you can always write new data and compute queries. In failure scenarios, queries will return results that don't incorporate previously written data. Eventually that data will be consistent and queries will incorporate that data into their computations.

This mitigates Brewer's proposal that revolves around reconciliation during a partition recovery phase. Marz' proposes no reconciliation, just a deferred sharing of the data.

[1] Martin Kleppmann's blog.
[2] Julian Browne's blog.

Architectural and Development Patterns

Event Sourcing
"[E]very operational command executed on any given Aggregate instance in the domain model will publish at least one Domain Event that describes the execution outcome. Each of the events is saved to an Event Store in the order in which it occurred. When each Aggregate is retrieved from its Repository, the instance is reconstituted by playing back the Events in the order in which they previously occurred... To avoid this bottleneck we can apply an optimization the uses Aggregate state snapshots." [1]

CQRS
Instead of having one data store for everything, "the change that CQRS introduces is to split that conceptual model into separate models for update and display, which it refers to as Command and Query respectively". [2]

I've seen this work well in a project that used a Lambda Architecture. Here, risk data was being written slowly but surely to a Hadoop cluster. When a day's data was complete, the data could be converted into a format that was digestible by the Risk Managers.

"Having separate models raises questions about how hard to keep those models consistent, which raises the likelihood of using eventual consistency." [2] For us this was fine as the Risk Managers were in bed while the munging was taking place.

"Interacting with the command-model naturally falls into commands or events, which meshes well with Event Sourcing." [2] And this is where Kafka came in.

Martin Fowler is cautious about using CQRS [2] but we had a good experience. In our domain, the Risk Managers were not interested in the Query model after two weeks and so it could be thrown away. Had it been needed at any time in the future, it could always have been generated again from the Command model.

Lambda Architecture
Lambda Architectures consist of three main pieces:

1. The batch layer, managing the master dataset (an immutable, append-only set of raw data) and pre-computing batch views.
2. The serving layer, indexing batch views so that they can be queried in a low-latency, ad-hoc way.
3. The speed layer, dealing with recent data only, and compensating for the high latency of the batch layer.

(taken from here).

As mentioned in the CQRS subsection above, "even if you were to lose all your serving layer datasets and speed layer datasets, you could reconstruct your application from the master dataset. This is because the batch views served by the serving layer are produced via functions on the master dataset, and since the speed layer is based only on recent data, it can construct itself within a few hours." [3]

Since Hadoop has an append-only, highly available file system (HDFS), it makes it an obvious choice for the batch layer.

"The serving layer is a specialized distributed database that loads in a batch Batch layer view and makes it possible to do random reads on it (see figure 1.9). When new batch views are available, the serving layer automatically swaps those in so that more up-to-date results are available. A serving layer database supports batch updates and random reads. Most notably, it doesn’t need to support random writes." [3] Cassandra was our choice for this layer.

Finally, "the speed layer only looks at recent data, whereas the batch layer looks at all the data at once. Another big difference is that in order to achieve the smallest latencies possible, the speed layer doesn’t look at all the new data at once. Instead, it updates the realtime views as it receives new data instead of recomput­ing the views from scratch like the batch layer does. The speed layer does incre­mental computation instead of the recomputation done in the batch layer." [3]

Lambda architecture shares some similarities with CQRS but is very different in that it never updates data (although it can add data that renders old data obsolete).

Feature Branching
The idea here is that each developer works on his own branch. He regular synchs with the main branch to lessen the pain of later merging his work into it. With Git, he'd regularly run:

> git checkout master
> git pull
> git checkout BRANCH_NAME
> git merge master
> git push

There is some debate on whether Feature Branching is an anti-pattern. Ideally, the system should be architected such that a feature is isolated to a particular silo of the code and not spill out to other components. Thus, a developer can merrily work on that particular silo and check into the main branch all the time. Mainly for political reasons, though, this is not always possible. Software like Stash that integrates with Jira can make code reviews even within a distributed team quite pleasant.

[1] Implementing Domain Driven Design, Vaughn Vernon
[2] Martin Fowler's blog.
[3] Big Data, Nathan Marz

Grid Computing vs. Distributed Cache Locking

If you want to change a piece of data in a distributed cache you could lock the data, change it then release the lock. But there is a more efficient way.

The idea is to leverage grid computing to execute a unit of work on the cache that does not require any locking. The advantage to this is that there are fewer network calls (one call to execute the work rather than the three to lock, mutate, unlock).

This architectural pattern works for two main reasons:

1. Key affinity - requests for a given piece of data get routed to the same server in the cluster.
2. One thread per key - although the host node may have a thread pool, only one of its threads can execute work against a particular key at any one time.

In Oracle's Coherence, this is achieved by implementing the EntryProcessor interface. "Within the process method, you don't need to worry about concurrency - Coherence guarantees that the individual entry processors against the same entry will execute atomically and in the order of arrival, which greatly simplifies the processing logic. It also guarantees that the processor will be executed in the case of server failure, by failing it over to the node that becomes the new owner of the entry it needs to process." - Oracle Coherence 3.5.