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]

Z-order

Z-ordering is an optimization technique in big data that allows faster access since similar data lives together. We discuss the algorithm that defines what is similar here. 

Imagine a logical grid where all the values of one column run across the top and all the values from another run down the side. If we were to sort this data, every datum can be placed somewhere in that grid.

Now, if the squares of the grid were mapped to files and all the data in each cell were to live in those files, we have made searching much easier as we now know the subset of files in which it may live. We've essentially sorted in not just one dimension but two (although we can do higher).

This can be especially useful when we want to sort the data but don't know exactly what to sort on - a common connundrum when dealing with events. Say we have event_time, insert_time and update_time. Which do we choose? We could sort the data three times, each time on one column but this is impractical with huge data sets. Enter z-order.

Note that the Z-Order really needs 2 or more columns on which to act. Only one column is the degenerate case. "Zorder/Hilbert etc on a single dimension are just a hierarchal sort" [Russell Spitzer on Slack].

(This is a good article about z-ordering from the perspective of Apache Iceberg.)

For an example in Delta Lake, we can see this code that creates a data set with columns c1, c2 and c3 whose values are [x, 99-x, x+50 mod 100] for x [0, 99]. After z-ordering it, these numbers are split into 4 different files. Generating a graphic illustrates how the data points are distributed over those files:

The idea behind how we calculate which cell a datum falls into is best described here on Wikipedia. But, in brief, the binary representation of the data points is interleaved to give a z-value per tuple. In our example, I see a [0, 99, 50] mapped to the byte array [0, 0, 0, 0, 0, 0, 0, 0, 0, 9, -112, 26].

I took a look at the Delta Lake code here where a Spark Column object is created that wraps a DL InterleaveBits type which in turn is a subclass of Spark's Catalyst Expression type. This executes on Spark's InternalRow, that is, the raw data on the executors.

The reason the code is doing this is to add a column with which we can repartition the data with the SQL CAST(interleavebits(rangepartitionid(c1), rangepartitionid(c2), rangepartitionid(c3)) AS STRING). The rangepartitionid keyword is part of the Delta Lake machinery.

Using this z-order value (plus a random key), the DataFrame the Delta code now calls repartitionByRange which samples the data [SO] and breaks it into discrete ranges.

Given the interleaving of the columns c1, c2 and c3 their order has minimal impact on the z-value so it's no surprise to see nearby data clustering into the same files, as we can see in the graphic. In fact, if you look at the DataFrame during the repartition process:

+---+---+---+-------------------------------------------+
| c1| c2| c3|c7b6b480-c678-4686-aa99-283988606159-rpKey1|
+---+---+---+-------------------------------------------+
|  0| 99| 50|                                       \t�|
|  1| 98| 51|                                       \t�|
|  2| 97| 52|                                       \t�b|
|  3| 96| 53|                                       \t�e|
|  4| 95| 54|                                       \b��|
|  5| 94| 55|                                       \b��|
|  6| 93| 56|                                       \b��|
|  7| 92| 57|                                       \b��|
|  8| 91| 58|                                       \b�|
|  9| 90| 59|                                       \b�|
| 10| 89| 60|                                       \b�b|
| 11| 88| 61|                                       \b�e|
| 12| 87| 62|                                       \b��|
| 13| 86| 63|                                       \b��|
| 14| 85| 64|                                       \f)�|
| 15| 84| 65|                                       \f)�|
| 16| 83| 66|                                       \f`|
| 17| 82| 67|                                       \f`|
| 18| 81| 68|                                       \f`b|
| 19| 80| 69|                                       \f`e|
+---+---+---+-------------------------------------------+

you can see the slowly changing values by which things are partitioned (column c7b6b480-c678-4686-aa99-283988606159-rpKey1 -  a random name so it doesn't clash with other column names. It's dropped immediately after the call to repartitionByRange)

Java and the GPU

Java was always built to abstract away the hardware on which it runs but it's approach to GPU has been somewhat late to the game [Gary Frost, YouTube].

There are projects out there that promise to give Java access to the GPU. I looked at Aparapi but it appears to be moribud. So, I gravitated to TornadoVM which Frost describes as "state of the art".

The trouble is that TornadoVM runs everything in a Docker image that has all the shared objects built in. This is fine for a quick demo - this is the result of running on my Quadro T2000:

docker-tornadovm$ ./run_nvidia_openjdk.sh tornado -cp example/target/example-1.0-SNAPSHOT.jar example.MatrixMultiplication
Computing MxM of 512x512
CPU Execution: 1.17 GFlops, Total time = 230 ms
GPU Execution: 268.44 GFlops, Total Time = 1 ms
Speedup: 230x

This demonstrates how the GPU runs a nested for-loop doing matrix multiplication much faster than the same code on the CPU. But it runs it all in a Docker container and I need to package a JAR everytime I make a change. How do I run it outside the container?

To work this out, I opened a shell in the Docker image and saw that the TornadoVM build it uses was built from Git branch d3062accc. So, the first thing was to checkout that branch of TornadoVM and build it.

I built with:

mvn clean install -Pgraal-jdk-11-plus

using the graalvm-ee-java11-22.3.4 JDK.

Note that you'll need Graal as the TornadoVM code has dependencies on it. I built my own Graal JDK by following the instructions here but using a different branch as I couldn't find the download for the graal.version defined in the TornadoVM pom.xml. Note, you'll also need mx and a bootstrapping JDK that has the right compiler interface (JVMCI), in my case labsjdk-ce-21.0.1-jvmci-23.1-b19.

So far, so good. I ran the tornado script which is just a wrapper around a call to the java executable (don't forget to set your JAVA_HOME environment variable to point at the Graal JDK) but it complained it could not see a tornado.backend file.

Again, a sneaky look at the Docker container indicated that we have to tell it which driver to use. So, I created the file and told it tornado.backends=opencl-backend but then tornado complained it didn't have the OpenCL drivers. Oops. 

You have to build the drivers you want seperately it seems. But if you try to build Tornado drivers without the native OpenCL dev library, you'll see:

TornadoVM/tornado-drivers/opencl-jni$ mvn clean install # yes, Maven cmake via cmake-maven-plugin
....
/usr/bin/ld: cannot find -lOpenCL
...

The Docker image saves you from having to install the OpenCL libraries on your machine. To get it working on bare metal, I played it safe and got an old Ubuntu box and installed them there. You'll need to install them with:

sudo apt install ocl-icd-opencl-dev

and then ran Maven in the opencl* sub directories. This time, the Maven build completed successfully.

However, running tornado in the subsequent dist folder still pukes but with something like:

Caused by: uk.ac.manchester.tornado.api.exceptions.TornadoRuntimeException: OpenCL JNI Library not found

at tornado.drivers.opencl@0.15.1/uk.ac.manchester.tornado.drivers.opencl.OpenCL.<clinit>(OpenCL.java:68)

... 11 more

Not what I was expecting. I found I needed to:

cp ./tornado-drivers/opencl-jni/target/linux-amd64-release/cmake/libtornado-opencl.so $TORNADO_SDK/lib

Where TORNADO_SDK is pointing at the relevent dist folder.

Now, finally, you can run on the bare metal:

$ tornado -cp target/classes/ example.MatrixMultiplication

Computing MxM of 512x512

CPU Execution: 1.21 GFlops, Total time = 222 ms

GPU Execution: 17.90 GFlops, Total Time = 15 ms

Speedup: 14x

(Results from an old NVIDIA GeForce GTX 650)

Note, you'll need to also run it with the Graal JVM. Set both the PATH and JAVA_HOME environment variables to point to it.

Where now?

This is a nice introduction to running Java on the GPU but it's just the start. There are many caveats. Example: what if your Java code throws an Exception? GPUs have no equivalent of exceptions so what happens then? More to come.

Dependency hell

In these days of ChatGPT, it's easy to forget that most of the time, a developer isn't actually cutting code at all, but debugging it. This is my own personal hell in getting Spark and Kafka in Docker containers talking to a driver on the host.

Firstly, I was seeing No TypeTag available when my code was trying to use the Spark Encoders. This SO answer helped. Basically, my code is Scala 3 and "Encoders.product[classa] is a Scala 2 thing. This method accepts an implicit TypeTag. There are no TypeTags in Scala 3". Yikes. This is probably one reason the upgrade path in Spark to Scala 3 is proving difficult. The solution I used was to create a SBT sub Project that was entirely Scala 2 and from here I called Spark.

The next problem was seeing my Spark jobs fail with:

Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 6) (172.30.0.7 executor 0): java.lang.ClassCastException: cannot assign instance of scala.collection.generic.DefaultSerializationProxy to field org.apache.spark.sql.execution.datasources.v2.DataSourceRDDPartition.inputPartitions of type scala.collection.immutable.Seq in instance of org.apache.spark.sql.execution.datasources.v2.DataSourceRDDPartition

This is a contender for the error message with the greatest misdirection. You think it's a serialization problem but it isn't directly so.

Although other Spark users have reported it, Ryan Blue mentions that it isn't really a Spark issue but a Scala issue.

Anyway, I tried all sorts of things like change my JDK (note the sun.* packages have been removed in later JDKs so you need to follow the advice in this SO answer). I tried creating an uber jar but was thwarted by duped dependencies [SO], Invalid signature file digest errors as some were signed [SO] that forced me to strip the signtures out [SO] but still falling foul of Kafka's DataSourceRegister file being stripped out [SO].

The first step in the right direction came from here and another SO question where the SparkSession is recommended to be built by adding .config("spark.jars", PATHS) where PATHS is a comma delimited string of the full paths of all the JARs you want to use. Surprisingly, this turned out to include Spark JARs themselves, including in my case spark-sql-kafka-0-10_2.13 which oddly does not come as part of the Spark installation. By adding them as spark.jars, they are uploaded into the work subdirectory of a Spark node.

After this, there was just some minor domain name mapping issues to clear up in both the host and container before the whole stack worked without any further errors being puked.

My "What data science can learn from software engineering" presentation

Dr Chris Monit and I presented this presentation at the London MLOps meetup last week. TL;DR: maximize your chances of a successful delivery in data science by adopting best practices that the software industry has established.

"Think of MLOps as the process of automating machine learning using DevOps methodologies" - Practical MLOps (O'Reilly)

Packaging Python

Python build tools are unifying behind a common interface of pyproject.toml.
This and this are great guides. The gist of the former is that you create a TOML file that conforms to a specification then you can use any build tool to run it. The gist of the latter is the whole Python packaging ecosystem.

The salient commands for building and deploying with your TOML file are:

python3 -m build
python3 -m twine upload --repository pypi dist/*

Note, you want to clean your dist directory first.

The Snag

The idea of using any Python build tool is not quite there yet. Poetry only implements a subset of the specification. Also, the specification has a leaky abstraction. On Discord, Prof. Nick Radcliffe explains that the promise of using "any" lead him to naively use setuptools.

Nick Radcliffe — 08/21/2023 2:37 PM

Also, in case anyone is interested (related to packaging, above) I'm currently in the process of packaging a fairly large Python codebase using new-style packaging (pyproject.toml rather than setup.py). It wasn't quite my first use of it, but this project is much more complex. Initially, I chose setuptools as the build backend, since (a) it didn't seem like it should matter much and (b) I didn't think I needed anything special. That was a big mistake for me: it turns out the setuptools back-end ignores almost everything except Python code in building your package. Whereas my package (which has over 10k files) also have about 1,000 non-python files (everything from .txt and .json to shape files, CSV files, and HTML and markdown and all sorts). Some of these are needed for testing (which for some reason some people think don't need to be distributed...as if people shouldn't care about whether the installed software works in situ, rather than just on the developer's machine in the CI system), but others are needed just in the ordinary course of using the software.  setuptools has a way to let you include extra stuff, but it's very manual and would be very error-prone for me. Anyway, the TL;DR is that I switched to Flit as the backend and everything "just worked". Not saying Flit will work better for you; but it sure as hell worked better for me!

Also, the reason I chose flit was that the third bullet in "Why use Flit?" is "Data files within a package directory are automatically included. Missing data files has been a common packaging mistake with other tools."

It also says: "The version number is taken from your package’s version attribute, so that always matches the version that tools like pip see." Which also seems extremely sane (and probably I don't need to do the automatic updating of my pyproject.toml to do that.

Success has many parents...

... but it appears that PyPI packages have only one. Although the authors tag can take a list, adding multiple entries is ignored. The reason is that it's best practise to use a mailing list (see here).

And so my package to facilitate the creation of synthetic data now lives in PyPI much like my Java code is deployed to mvnrepository.