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".

Avoiding Spark OOMEs

Spark can process more data than it can fit into memory. So why does it sometimes fail with OutOfMemoryExceptions when joining unskewed data sets?

An interesting way to counter OOMEs in a large join is here [SO] where rows are given a random integer seed that is used in addition to the usual condition. In theory, this breaks down the data into more manageable chunks.

Another standard exercise is to repartition the data. But this causes a shuffle and it may actually be the repartition itself that causes of an OOME.

In practice, I've found persisting the data frame to disk and reading it back yields better results. The number of partitions being written is rarely the number that is read back. That is, you get a more natural partition for free (or almost free. Obviously, some time is taken in writing to disk). And there is no repartition that could throw an OOME.

This question came up on Discord where somebody is trying to crossJoin a huge amount of data. I suggested a solution that uses mapPartitions. The nice thing about this method is that your code is passed a lazy data structure. As long as you don't try to call something like toList on it, it will pull data into memory as needed and garbage collect it after it's written out.

By using a lazy Iterator, Spark can write far more memory than it has to disk. As Spark consumes from the Iterator, it measures its memory. When it starts looking a bit full, it flushes to disk. Here is the memory usage of this code that uses mapPartitions to write to /tmp/results_parquet a data set that is much larger than the JVMs heap:

Spark with 0.5gb heap writing 1.3gb files

If we run:

watch "du -sh /tmp/results_parquet"

we can see that upon each GC, more is written to disk.

The result is a huge dataframe that could not fit into memory can now be joined with another.

As an aside: Uber has been doing some work on dealing with OOMEs in Spark. See their article here. TL;DR; they're proposing that in the event of an OOME, Spark adapts and increases the memory to CPU ratio by asking come cores to step down before it re-attempts the failed stage. Ergo, each compute unit has more memory than before. 

Hilbert Curves

When you want to cluster data together over multiple dimensions, you can use Z-Order. But a better algorithm is the Hilbert Curve, a fractal that makes a best attempt to keep adjacent points together in a 1-dimensional space.

From DataBrick's Liquid Cluster design doc we get this graphical representation of what it looks like:

Dotted line squares represent files

A Hilbert curve has the property that adjacent nodes (on the red line, above) have a distance of 1. Note that a property of the Hilbert curve is the adjacent points on the curve are nearest neighbours in the original n-dimensional space but the opposite is not necessarily true. Not all nearest neighbours in the n-dimensional space are adjacent on the curve. How could they be if points have more than 2 neighbours in the original space?

An algorithm in C for navigating this square can be found here. A Python toolkit for handling Hilbert curves can be found here [GitHub]. And a Java implementation can be found here [SO].

The application of this in Big Data is that the data is now sorted. If we were to read through the files following the red line, then each node we encountered is one away from the last. Z-Ordering does not have this property.

Z-ordering. Lines indicate contiguous data. Colours indicate different files.

Unlike the Hilbert curve at the top of this page, there are some large jumps. In fact, the average step is not 1.0 as for the Hilbert curve but 1.557 in this example - over 50% more!

This greater efficiency is true even if we don't take the unlikely case that the data is tightly packed. Below are examples where the data is more realistic and not every possible point (a red +) is actually associated with data (a blue circle).

A Hilbert curve over sparse data

To understand what is going on, we need to appreciate Gray Codes [Wikipedia] which is an alternative numbering system in binary where adjacent numbers only differ by one bit changing (see that parallel with Hilbert curves?). For each bit, for each dimension, we create a mask from the Gray code and do some bit manipulation found here and we'll eventually have a bijective map ℤ^d → ℤ.

The jumps between adjacent data points is less extreme in Hilbert curves. You can see this by-eye if look at a slightly larger space (code here):

A Hilbert curve over sparse data

Typically, the jumps between data points are never more than a couple of positions (average of 1.433). Now, compare this to a similar space using Z-Ordering:

Z-Order over a similar sparse space

and you can see larger jumps between some data points. The average is 2.083 in this run. That's 45% higher than in the Hilbert curve.

Hilbert curves are not currently implemented in Apache Iceberg but are in Databrick's Delta Lake.

GPU vs CPU vs AVX

Vector databases are all the rage. So, I looked at three different ways of multiplying vectors: CPU, GPU and Advanced Vector Extensions that leverages SIMD instructions if your hardware supports them. To access the GPU, I'm using the Tornado Java VM. For AVX, I'm using the JVM's jdk.incubator.vector module, available since JDK16.

(Code in my GitHub repo here).

The reason we're looking at vector mulitplication is that searching for vectors (what the vector DB is all about) usually uses something like the approximate nearest neighbour algorithm. One way to implement it is something like Ethan Lui's implementation mentioned in a past blogpost here. Briefly: it multiplies your vector by random vectors resulting in a vector whose bits are on or off depending on the sign of each element in the product.

The results are as follow (note, the GPU is a Quadro T2000 that apparently has 4gb of memory, 1024 cores and a bandwidth of 128 gigabits per second).

You can see that there is a huge fixed cost to using the GPU but once you get sufficiently large vectors, it's worth it. But what causes this fixed cost?

On my Intel Xeon E-2286M  CPU @ 2.40GHz, kernel calls take typically 17.8ns.

  17.776 ±(99.9%) 0.229 ns/op [Average]

  (min, avg, max) = (17.462, 17.776, 19.040), stdev = 0.306

  CI (99.9%): [17.547, 18.005] (assumes normal distribution)

JNI calls take a little longer at about 21.9ns:

  21.853 ±(99.9%) 0.488 ns/op [Average]

  (min, avg, max) = (21.345, 21.853, 23.254), stdev = 0.651

  CI (99.9%): [21.365, 22.340] (assumes normal distribution)

So, it doesn't seem that the fixed costs incurred in the GPU vector multiplication is due to context switching when calling the kernel or calls via JNI.

Note the maximum vector size for this test was 8 388 608 floats. 

That's 268 435 456 bits or 0.25 gigabits.

Based on just bandwidth alone and ignoring everything else, each call should be about 1.95ms. This matches the average observed time (1.94971ms). 

This suggests the actual calculation is incredibly fast and only the low bandwidth is slowing it down. Tornado VM appears to have minimal room for improvement - you really are getting the best you can out of the hardware.

Cloud native

A cloud native approach to writing code is that the instance in which it lives can die at any time.

"Users sometimes explicitly send the SIGKILL signal to a process using kill -KILL or kill -9. However, this is generally a mistak. A well-designed application will have a handler for SIGTERM that causes the application to exit gracefully, cleaning up temporary files and realeasing other resources beforehand. Killing a process with SIGKILL bypasses the SIGTERM handler." - The Linux Programming Interface (Micahel Kerrisk)

Using docker stop sends SIGTERM.
Using docker kill sends SIGKILL.

The latter does not give the JVM a chance to clean up. In fact, no process in any language has the chance to clean up with SIGKILL. (SIGTERM on any thread - not just main - causes the whole JVM process to end and shutdown hooks to execute.) 

A Tini problem...

If the JVM process creates another process is killed with SIGKILL, that process carries on living but its parent becomes (on Ubuntu 20.04.6 LTS) systemd which in turn is owned by init (PID 1).

Running your JVM directly in a Docker container has some issues. This revolves around Linux treating PID 1 as special. And the ENTRYPOINT for any Docker container is PID 1.

In Linux, PID 1 should be init. On my Linux machine, I see:

$ ps -ef | head -2
UID        PID  PPID  C STIME TTY          TIME CMD
root         1     0  0 Oct21 ?        00:18:23 /sbin/init splash

This process serves a special purpose. It handles SIGnals and zombie processes. Java is not built with that in mind so it's best to bootstrap it with a small process called tini. There's a good discussion why this is important here on GitHub. Basically, Tini will forward the signal that killed the JVM onto any zombies that are left behind. This gives them the chance to clean up too. 

It also passes the JVM's exit code on so we can know how it failed. Exit codes 0-127 are reserved [SO] and the value of the kill (kill -l lists them) is added to 128. If you want to set the exit code in the shutdown hook, note you need to call Runtime.halt rather than Runtime.exit (to which System.exit delegates). The exit method will cause the JVM to hang in this situation [SO].

ML and Logs (pt2)

Further to my attempt to use machine learning to make sense of huge amounts of logs, I've been looking at the results. My PoC can:

Find log entries with the highest information

When debugging my Kafka cluster, these lines had the highest average entropy:

kafka1: 2023-07-04 14:14:18,861 [RaftManager id=1] Connection to node 3 (kafka3/172.31.0.4:9098) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

kafka1: 2023-07-04 14:17:32,605 [RaftManager id=1] Connection to node 2 (kafka2/172.31.0.3:9098) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

kafka2: 2023-07-04 14:17:31,957 [TransactionCoordinator id=2] Connection to node 3 (localhost/127.0.0.1:9093) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

kafka1: 2023-07-04 14:17:32,605 [RaftManager id=1] Node 2 disconnected. (org.apache.kafka.clients.NetworkClient)

kafka2: 2023-07-04 14:17:31,957 [TransactionCoordinator id=2] Node 3 disconnected. (org.apache.kafka.clients.NetworkClient)

As it happened, this correctly highlighted my problem (Docker Compose networking was misconfigured). But I don't know if I got lucky.

Bucket similar-but-different lines

Using the same algorithm as Twitter, we can bucket similar but lexically different lines, for example:

2023-07-04 14:14:21,480 [QuorumController id=3] ConfigResource(type=TOPIC, name='__consumer_offsets'): set configuration cleanup.policy to compact (org.apache.kafka.controller.ConfigurationControlManager)

2023-07-04 14:14:21,489 [QuorumController id=3] ConfigResource(type=TOPIC, name='__consumer_offsets'): set configuration compression.type to producer (org.apache.kafka.controller.ConfigurationControlManager)

This means that we can:

• discard boilerplate lines of little value like those above
• check the distribution of all nodes in a given bucket (for example, if one node is under-represented within a bucket - that is, not logging the same as its peers - this might be an issue).

There's one slight gotcha here: in the Kafka example above, we're using the Raft protocol so it's not too surprising that the number of nodes is N-1 for some configurations as one has been elected leader and the others are followers.

Trace high information tokens through the system

Words with high entropy can be traced across my cluster. For instance, my PoC classified wUi1RthMRPabI8rHS_Snig as possessing high information. This happens to be an internal Kafka UUID for a topic and tracing its occurrence through the logs show that despite Docker network issues, all nodes agreed on the topic ID as did the client. So, clearly some communication was happening despite the misconfiguration.

Investigation

I finally solved my Kafka problem. The Kafka client was running on the host OS and could see the individual Kafka containers but these brokers could not talk to each other. The reason was they needed to advertise themselves both as localhost (for the sake of the Kafka client that lives outside Docker) and also using their internal names (so they could talk within the Docker network).

My PoC could not tell me exactly what the problem was but it successfully highlighted the suspects.

The PoC

So, how does the PoC work? For the entropy, we train the model on a dictionary of English words so it can learn what is a "normal" word, rather than say wUi1RthMRPabI8rHS_Snig. We disregard lines that are fewer than 6 words (including the FQN of the classes - each package being one word); take the average entropy and present the lines that look the most informative.

For the LSH, we use one-hot encoding of word shingles to create our vectors.

Future plans

I'd like to show the graph of paths the high-entropy words take through the system (node and log line).

I'd also like to try other systems. Maybe I got lucky with Kafka as there are lovely, high-entropy UUID scattered throughout the logs (for example, consumer group IDs).

Thirdly, this PoC has been great for small amounts of data, but what about big data? It really needs to be rewritten in a JVM language and made to run in Spark.

Memories are made of these

Some notes on new memory models I've been looking at recently.

Zero copy

"Device controllers cannot do DMA directly into user space, but the same effect is achievable by exploiting ... [the fact] more than one virtual address can refer to the same physical memory location. [Thus] the DMA hardware (which can access only physical memory addresses) can fill a buffer that is simultaneously visible to both the kernel and a user space process." - Java NIO, Ron Hitchens

Virtual memory paging is "often referred to as swapping, though true swapping is done at the process level, not the page level" [ibid].

An excellent visual representation of what's going on in during a zero-copy is here from Stanislav Kozlovski (who ends with the knock-out punch that is makes very little difference to Kafka since generally costs of network IO and encryption cancel any savings). Anyway, the take-away points are: 

• Zero-copy "doesn’t actually mean you make literally zero copies" it's just that it "does not make unnecessary copies of the data."
• Fewer context switches happen.
• A further optimization to DMA is where the disk "read buffer directly copies data to the NIC buffer - not to the socket buffer.  This is the so-called scatter-gather operation (a.k.a Vectorized I/O).  [It is] the act of only storing read buffer pointers in the socket buffer, and having the DMA engine read those addresses directly from memory."

Java's new vector API

A new way of dealing with vectors is outlined at JEP426 (Vector API). It leverages new CPU features like Advanced Vector Extensions [Wikipedia] that provide new machine instructions to execute Single Instructions on Multiple Data (SIMD).  

Martin Stypinski has an interseting article that shows adding two floating point vectors together gain very little from the new API but a linear equation like y = mx + c (which has obvious applications to machine learning) can improve performance by an order of magnitude.

Project Panama

Project Panama deals with interconnecting the JVM with native code. Oracle's Gary Frost talks about this in his presentation on accessing the GPU from Java. The difficulty he encountered was allocating heap memory and passing it to the GPU. Unfortunately, the garbage collector might reorganise the heap making the pointer to that memory obsolete. With Project Panama, this would not happen as the allocation would be through the JVM but off the heap. 

Apache Arrow

Arrow provides an agreed memory format for data so you can "share data across languages and processes." [docs]

This differs from Google's Protobuf in that "Protobuf is designed to create a common on the wire or disk format for data." [SO] Any data from Protobuf that is deserialized will be done in the the same way that language always handles it.

This inter-process ability allows Spark (which runs in the JVM) to use Pandas (which runs in a Python process).

"Perhaps the single biggest memory management problem with pandas is the requirement that data must be loaded completely into RAM to be processed... Arrow serialization design provides a “data header” which describes the exact locations and sizes of all the memory buffers for all the columns in a table. This means you can memory map huge, bigger-than-RAM datasets and evaluate pandas-style algorithms on them in-place without loading them into memory like you have to with pandas now. You could read 1 megabyte from the middle of a 1 terabyte table, and you only pay the cost of performing those random reads totalling 1 megabyte... Arrow’s memory-mapping capability also allows multiple processes to work with the same large dataset without moving it or copying it in any way. "[10 Things I hate about Pandas, by Pandas author, Wes McKinny]

"The ability to memory map files allows you to treat file data on disk as if it was in memory. This exploits the virtual memory capabilities of the operating system to dynamically cache file content without committing memory resources to hold a copy of the file." [NIO - Hitchens].

MySQL vs Postgres

There's a great comparison between the two major open source DBs here at Uber. Amongst the many insights, there is a mention that MySQL uses a cache "logically similar to the Linux page cache but implemented in userspace... It results in fewer context switches. Data accessed via the InnoDB buffer pool doesn’t require any user/kernel context switches. The worst case behavior is the occurrence of a TLB [Translation Lookaside Buffer] miss, which is relatively cheap and can be minimized by using huge pages."

"On systems that have large amounds of memory and where applications require large blocks of memory, using huge pages reduces the number of entries required in the hardware memory management unit's translation look-aside buffer (TLB). This is beneficial because entries in the TLB are usually a scarce resource... For example, x86-32 allows 4mb pages as an alternative to 4kb pages)" [The Linux Programming Interface]