Merkle Trees

Merkle Trees are a data structure that allows for efficient and secure validation of contents. They are typically implemented as binary trees. Given N pieces of data, they would have 2N nodes and log(N) height. Each leaf node is the hash of a piece of data and every non-leaf node is a hash of its children's hashes. With only a hash stored at each node Merkle trees have a small, predictable size compared to the data of a large N.

Examples of Merkle tree usage include detecting data inconsistencies between replicas in NoSQL databases, ensuring file integrity in distributed storage systems, and verifying blockchain transactions. We can easily create one with pymerkle.

└─9f42c047...

    ├──50fcd75a...

    │   ├──4c4b77fe...

    │   │   ├──e8bcd97e...

    │   │   │   ├──2215e8ac...

    │   │   │   └──fa61e3de...

    │   │   └──9c769ac2...

    │   │       ├──906c5d24...

    │   │       └──11e1f558...

    │   └──fed7af7d...

    │       ├──2b15ae18...

    │       │   ├──53304f5e...

    │       │   └──3bf9c81c...

    │       └──8007dd69...

    │           ├──797427cf...

    │           └──195f58bc...

    └──555da077...

        ├──85224a5c...

        └──5c889ef4...

A "Merkle proof" is a list of intermediate hashes from the path between a leaf node representing the data you want to prove and the root of the tree. Generating the proof is like a modified DFS. The beauty of this is that anyone can verify the data is included in the tree without the whole tree being revealed. 

{

    "metadata": {

        "algorithm": "sha256",

        "security": true,

        "size": 10

    },

    "rule": [

        0,

        0,

        1,

        0,

        0

    ],

    "subset": [],

    "path": [

        "53304f5e3fd4bcd20b39abdef2fe118031cc5ae8217bcea008dea7e27869348a",

        "3bf9c81c231cae70b678d3f3038f9f4f6d6b9d7adcf9b378f25919ae53d17686",

        "8007dd69b92a67ea6410098635fa8ba53c44a5994c7e5d92b99e27f0711c626f",

        "4c4b77fe3fc6cfb92e4d3c90b5ade42f059a1f112a49827f07edbb7bd4540e7b",

        "555da077fcadba1f23e0f2bfac8793e6a3c79a0d605902df34ab43d3e0fb487c"

    ]

}

Verifying the proof requires only calculating the root hash from the provided proof and the data being verified. If the calculated root matches the known root of the tree then the data is present in the tree. We are using less space and less compute than if we were iterating a list to check if data is present.

That was an inclusion or audit proof. We can also do a consistency proof. In an append-only tree we can verify earlier versions of the tree against later versions to make sure no tampering has occurred. The later version must include everything in the earlier version, in the same order, and all new entries come after old entries.

{

    "metadata": {

        "algorithm": "sha256",

        "security": true,

        "size": 10

    },

    "rule": [

        1,

        0,

        0,

        0,

        0

    ],

    "subset": [

        0,

        1,

        0,

        0,

        0

    ],

    "path": [

        "fa61e3dec3439589f4784c893bf321d0084f04c572c7af2b68e3f3360a35b486",

        "2215e8ac4e2b871c2a48189e79738c956c081e23ac2f2415bf77da199dfd920c",

        "9c769ac26f8d61ff40859e5201537845555136f0fd7ab604f7033180fbe76af9",

        "fed7af7d64bf0a73fcad018df1219928dbafa4d96b5d78f8a5e9be66ff0ada38",

        "555da077fcadba1f23e0f2bfac8793e6a3c79a0d605902df34ab43d3e0fb487c"

    ]

}

Given how powerful and pervasive Merkle trees are, I'm surprised they aren't discussed more along with other common data structures. It seems that with AI-generated content making verifiable content more imperative, their usage will only increase going forwards.

ML at Scale Part 3: distributed compute

In part 2 I focused on ML when your data won't fit in memory. This post will move on to slow, or compute bound ML instead. I'll continue to use Dask and explore how it can help us.

Dask leverages multiple CPU cores to enable efficient parallel computation on a single machine. It can also run on a thousand-machine cluster, breaking up computations and routing them efficiently. The Comparison to Spark documentation is a great reference for understanding Dask in the context of an older tool and that older post. Perhaps the most interesting difference is Spark just being an extension of the MapReduce paradigm and Dask being able to implement more sophisticated algorithms by being generic task scheduling-based.

Sticking with the scikit-learn examples, here is replacing a parallel algorithm's Joblib backend with Dask to (potentially) spread work out across a cluster.

The Dask documentation is quick to point out in their best practices that not everyone needs distributed ML as it has some overhead. Compiling code with Numba or Cython could help, as could intelligently sampling some of your data. In this post I got huge speedups by vectorizing some code that was looping through large matrices.

Across this three part series we've now seen how to speed up reading large datasets, work with datasets that don't fit entirely in memory, and distribute processing across multiple machines. There's obviously a lot more to this, but I wanted to develop a better intuition for how to approach these types of issues and at least know where to start. Hopefully you learned something too.

UPDATE:

A specific tool isn't supposed to be the focus of this post. Dask was used here to illustrate the idea and show how simple it can be, but there are other options in this space. Here are a couple of other examples that I've come across:

• tune-sklearn, which is part of the Ray framework for building distributed applications, can do distributed hyperparameter tuning
• Distributed training with TensorFlow shows how you can distribute model training within the TensorFlow ecosystem

UPDATE 2:

Both Pandas 2.0 and Polars now use Apache Arrow as a memory model. Polars, a relatively new entrant to this space, is implemented in Rust and exposes a Python API. It is created specifically for fast data processing, not ML, but overlaps enough with this series that it is worth checking out.

UPDATE 3:

Since I mentioned Numba earlier, I'll also mention Jax another, newer library in that space. Jax has a NumPy-like interface, works with GPUs, offers JIT compilation, and supports automatic differentiation, vectorization, and parallelization. Check out their simple NN example to see it in action. It's not replacing distributed compute or even competing with Dask, the point is more that there are now a lot of amazing tools that make working with data easier and faster.

ML at Scale Part 2: memory

When data we want to train a machine learning model on becomes too big to fit in memory, we need to find a way to work on subsets of the data. I hinted at this in part 1. We can use Pandas to read chunks of a file, but that is fairly primitive and slow.

Libraries like Vaex and Dask attempt to abstract this away. 

Vaex provides lazy, out-of-core (not all in memory at once) DataFrames via memory-mapping. Pre-processing and feature engineering are more efficient, and memory is freed up for model training. It also has a vaex.ml package which provides a scikit-learn wrapper.

Dask provides large, parallel DataFrames composed of smaller Pandas DataFrames. This helps with data too big to fit in memory because the individual Pandas DataFrames can be stored on disk. DaskML provides estimators designed to work with Dask DataFrames.

The accuracies both came out to 96%. Similar ideas, different implementation. In fact, these DataFrames remind me a little bit of the persistent data structures covered in my Exploring Immutability post.

Even with these fancy DataFrames, many machine learning algorithms are designed train on all the data at once. If our data is too big to fit in memory then that's going to be a problem. As in the code above, we need to use online learning or incremental algorithms to solve this problem. The Incremental and IncrementalPredictor classes handle "streaming" the data (in batches) and we specify upfront the possible classes.

DaskML adds several generalized linear model implementations. scikit-multiflow is designed for actual streaming data and adds several other online learning algorithms. Neural networks are also trained in this manner, often being fed "mini-batches", so they are good candidates for datasets that don't fit in memory as well.

Stay tuned for part 3 where I'll get into being compute-constrained instead of memory-constrained.

UPDATE:

There's a lot of I/O and memory-related work coming out of the TensorFlow community as well. Checkout Better performance with the tf.data API as it overlaps nicely with Parts 1 and 2 of this series.

ML at Scale Part 1: I/O

I recently came across a somewhat large dataset from a Kaggle competition where the data was provided as an approximately 6 GB CSV file. A frequent comment in the discussion forum was how long it took just to read this file. A few GBs is large enough where this starts to become noticeable, but it's not really that big. If a laptop can have a 1+ TB drive and 32 GB memory then this isn't even in the realm of "big data". That's good, though, because it means there are some simple tricks we can use to cut down on that read time.

The pandas read_csv() method takes about 63 seconds for me. That's our baseline.

First we try reducing the precision. This gets us to 59 seconds. Not great.

Next we try reading the files in chunks. This is actually slower, but the technique could help us if the file didn't fit entirely in memory. More on that in future posts.

Then we try Dask which will spread the work across multiple processers. 30 seconds. Better, but I still don't want to wait that long. More on Dask in future posts as well.

Finally we convert the CSV file to a different format. I tried Apache Parquet but there are others. It's a binary columnar format (remember Column-oriented Database Basics?). Stored in this manner the data is 2.5 GB. And this gets us to just 3 seconds for reading the whole file!

Converting our data to the binary file format and possibly reducing the precision or using Dask as well would really shorten our feedback loop while training a ML model. It would seem that any cleaning of the data or preprocessing that we can do ahead of time would make sense to do once, before converting the file format, when the data is this size.

UPDATE:

Apache Arrow is another project to checkout in this space along with memory-mapped files. Reading this data in the Feather file format is even faster than Parquet.

Column-oriented Database Basics

This is a short post on column-oriented databases. I'll barely scratch the surface, but among the types of NoSQL databases—document, key-value, column-oriented and graph—I've always thought column-oriented was the most difficult to wrap my head around. Hopefully we can get past that initial hurdle here, run a few queries, and they will seem like less of a mystery.

A relational database is optimized for retrieving rows of data. This works well for transactional applications. A column-oriented database is optimized for retrieving columns of data. This works well for analytical applications and some queries, like aggregations, become really fast because much less data needs to be read from disk to retrieve the whole column. There seems to be a lot of overlap between column-oriented databases and data warehouses.

A relational database would store 3 rows of data like this:

1:a,b,c;2:d,e,f;3:g,h,i

While a column-oriented database would store the same data like this:

a:1,d:2,g:3;b:1,e:2,h:3;c:1,f:2,i:3

For a low-cardinality column, compression algorithms work very well. Something like a:2,a:3 becomes a:2,3. In some ways this is like normalizing a relational database to reduce data duplication, but I don't think that enables the same level of compression or gives you the same data locality benefits.

Of course column-oriented databases aren't good for all workloads. They aren't optimized for queries that touch many fields. Writes can also be slow since they aren't just appending to the end of a file.

So where do they fit into a big data architecture? When thinking about this I remembered the article How to Move Beyond a Monolithic Data Lake to a Distributed Data Mesh. It avoids mentioning specific products, but they way I interpret it is that their first generation was dedicated data warehouse products where you did ETL. The second generation was more ELT, maybe using Hadoop and Spark. The third generation, the eponymous data mesh, unifies batch and stream processing, perhaps adding Kafka to the above mix. Using the CQRS pattern, for example, a column-oriented database could fulfill some of the Q (query) operations as a sort of data warehouse.

For my first column-oriented database adventure, after that background research, I chose Amazon Redshift and I followed their Getting Started with Amazon Redshift guide. It only took about an hour to spin up a Redshift cluster, upload their sample data to an S3 bucket, copy that into Redshift, then run a few SQL queries.

The big surprise and takeaway was that this felt just like using a relational database. The data I uploaded was in text files (basically CSV files) and was used to create tables. The queries I ran were SQL queries that joined tables, selected different fields and aggregated the results. The details of how the database works under the covers is interesting, but doesn't inform its usage. There isn't much mystery after all.

SELECT * FROM event;

SELECT firstname, lastname, total_quantity FROM (SELECT buyerid, sum(qtysold) total_quantity FROM sales GROUP BY buyerid ORDER BY total_quantity desc limit 10) Q, users WHERE Q.buyerid = userid ORDER BY Q.total_quantity desc;

A couple of other column-oriented databases you may have heard of are Cassandra and HBase. NoSQL databases are built to scale horizontally so you also need to consider how the CAP Theorem applies to your situation when choosing among them as they each make different trade-offs in addition to their unique features. The best choice will be highly data and workload specific.

Eigenvectors and PageRank

Eigenvectors are another topic that I used in a math class or two but never really understood what they were for. As I mentioned in my last post Bipartite Graphs and Google Adwords, I like how the professors of the Coursera class Mining Massive Datasets gave great examples of how to apply some academic concepts to real world problems. Their PageRank material is no different as it explains how eigenvectors are used in the algorithm. Here is my summary of the class' link analysis chapter which I'm sure is greatly simplified from what Google really does, but I think it's interesting nonetheless.

Early search engines used an inverted index which I covered in my tf-idf post. That approach is vulnerable to term spam. People would put invisible text on their pages (same color as the background) to trick web crawlers into thinking that the page was about a topic it wasn't. The PageRank algorithm solves that problem by giving more importance to pages that have many in-links, and especially important in-links. Even if someone creates millions of fake pages that link to their real page, it's importance will not be that high because the fake pages will have a low importance as no one is linking to them.

The web can be thought of as a directed graph where web pages are the vertexes and links are the arcs. The basic idea of PageRank is that you simulate web surfers starting at random spots in the web graph and then following random out-links. The most important pages, the pages with the most in-links, will end up with the most surfers.

To implement this simulation, consider a transition matrix M which has n rows and columns where n is the number of pages crawled. M_ij has the value 1/k if page j has k arcs out and one of them is to page i, otherwise it has the value 0. Start with a vector v₀ which has all elements set to 1/n. The first step is to multiply v₀ by M, the second step is to multiple by M², and so on (this is an example of a Markov process which means you can make predictions on its future state based only on the present state and do not need the history).

If the web graph is one strongly connected component and doesn't have any dead ends, then v = Mv is the limiting distribution. The limit would be reached when multiplying the distribution by M another time doesn't change the distribution. Vector v is an eigenvector of M because v = λMv. Vector v is also the principal eigenvector because M is a stochastic matrix and the eigenvalue associated with the principal eigenvector is 1. In practice it takes 50-75 iterations to reach this limit.

M is too big for us to use Gaussian elimination. M is also very sparse, so it makes sense to store only the non-zero elements. An entire column can be represented by the out-degree of a page and the row number of the non-zero elements (since those element's values will be 1 divided by the out-degree). That means we only need to store a little over 4 bytes for each non-zero element. Then we can solve for v using MapReduce and v tells us the "page rank" of each page.

In reality the web is not strongly connected. There are dead ends and there are spider traps, or cycles, in the graph. This problem is solved by the concept of taxation. Each surfer has a small probability of teleporting to a random page instead of following an out-link. There are also other algorithm variations for dealing with link spam. For example, besides page content, you also consider the link text or words near the link, so you are getting other people's take on what the page is about instead of relying solely on the page owner.

UPDATE:

This same idea can be applied to sports ratings where links are replaced by something like goals scored. My Machine Learning for NCAA Basketball Prediction - Performance Edition post has more details and some code.

Resources:
http://infolab.stanford.edu/~ullman/mmds/ch5.pdf

Bipartite Graphs and Google Adwords

Continuing my series of posts relating to the Coursera Mining Massive Datasets class, this post summarizes the chapter on Google AdWords. I liked how the professors showed it to be related to matching in a bipartite graph--something that I learned about in a university graph theory class but without such a practical context as web advertising. I think that too often happens, at least in my experience. I've taken a lot of math classes (and done well in them), but I still don't feel like I had much practice applying what I learned to real-world problems.

To be clear, AdWords is for advertisers and AdSense is for website owners. Google provides an explanation in The difference between AdWords and AdSense. Today I'm focusing on how AdWords works.

The "adwords" model matches web searches with advertisements. Advertisers bid to be shown in response to certain search queries. If a user clicks on one of the ads shown to them then the advertiser will pay. Of course, the most relevant ads are clicked more often than less attractive ads, so the challenge is displaying the ads with the highest bids that are most likely to be clicked, while staying within an advertisers budget. This is a problem that traditional newspapers and magazines don't really have. They can only target specific niches of people (like people who buy Golf Digest), but web advertisers can target individuals (like people searching for a specific brand of golf clubs).

Google knows the click-through rate, or the percentage of the time an ad is clicked when it is displayed. It knows how much of an advertiser's budget has been spent. Google also knows what people have searched for in the past, but it doesn't know for sure what people are going to search for in the future. The "adwords" problem therefore needs a greedy algorithm because all you can do is make the best choice for each search and hope it results in the best overall outcome.

A simplified version of this "adwords" problem can be modeled as maximal matching in bipartite graphs. A matching is a subset of edges where no vertex is an end of two or more edges. A perfect matching contains every vertex and a maximal matching is the largest possible matching for the graph. In this case one side of the graph is search queries, the other side is ads, and the edges are who the ads could be shown to. The maximal matching is the best way to display the ads. The simplifying assumptions are that one ad is shown, all advertisers have the same budget, click-through rates are the same, and bids are either 0 or 1.

The obvious greedy algorithm for matching will consider edges in the order they are given. An edge is part of the matching if neither end is connected to an edge already added to the matching. The competitive ratio is defined as the ratio between the worst online algorithm and the best offline algorithm. The offline solution is the optimal solution because all information about the problem is known in advance. For our bipartite matching solution the ratio is only 0.5, that is it will always find at least half as many matches as what is optimal.

It is possible to do better. The more realistic BALANCE algorithm considers the highest bidder and the highest remaining budget. By doing so it results in a competitive ratio of 0.63 which is the highest possible for an online algorithm.

UPDATE:

Bipartite matching implementations can be found in NetworkX and SciPy.

Resources
http://infolab.stanford.edu/~ullman/mmds/ch8.pdf

Locality Sensitive Hashing

In my previous post on tf-idf, I summarized my notes on how to look for similar documents in the sense that the documents are about the same topic. However, there is a slightly different problem of finding similar documents that are really the same document. They could be, for example, plagiarized, a mirror website, or a news story from the same source carried by multiple news outlets. When there are too many pairs of documents to efficiently compare them all, we can use a technique called locality sensitive hashing or LSH. Today I'll summarize my LSH notes.

In order to find these lexically similar documents, you need to represent each document as a set of sentences or phrases. One technique for this is called shingling and a k-shingle is any substring with length k in the document. The value of k needs to be large enough that the probability of a shingle appearing in any given document is low. To represent shingles more concisely, use a hash function to hash the substrings to a number. Then, for example, each shingle is represented as four bytes instead of a k byte substring.

Even with four byte shingles the set for a document still takes up more space than the document itself. We need to replace these large sets with smaller representations called signatures. A signature is the result of several hundred minhash calculations. Consider a matrix where there is a row for each shingle and a column for every document sets. The matrix has a 1 for a particular row and column if that shingle appears in the set. Pick a permutation of rows and then the minhash value for a column is the number of the first row (in permuted order) with a 1 in that column.

In practice, storing the sparse matrix and permuting and sorting millions of rows is too time-consuming, but it can be simulated by random hash functions that map row numbers to as many buckets as there are rows (and ignoring the small number of collisions). Instead of picking n random permutations you pick n randomly chosen hash functions. If a column has a 1 for row r, then the corresponding signature matrix column is updated with the bucket numbers as long as they are less than the current values.

From the signature matrix we can estimate the Jaccard similarities of the underlying sets. Mind = blown.

It still might not be possible to efficiently check the similarity of all pairs of documents because there are just too many pairs. To find the most similar pairs or all pairs above a certain similarity threshold, we need to focus only on pairs that are likely to be similar. The LSH approach here is to hash items several times and hope that similar pairs will end up in the same bucket at least once. Any pair that hashed to the same bucket in any of the hashings is a candidate pair (even though some of them are false positives).

Divide the signature matrix into b bands where each band consists of r rows. Hash each vector (the portion of each column in a band) to a large number of buckets. Each band has its own buckets. The similarity threshold is an S-curve and is approximately:

For a Jaccard similarity s, the probability of a false negative, i.e. missing a candidate pair, is given by:

Finally, examine the candidate pairs to determine if they are above the previously determined similarity threshold.

Besides the minhash family of functions, there are also other function families that apply to other distance measures like Hamming distance or edit distance. LSH works well for fingerprint matching and entity-resolution.

UPDATE:

Spark implements several LSH operations and has more info on how they can be used.


Resources:
http://infolab.stanford.edu/~ullman/mmds/ch3.pdf

Bloom Filters

A bloom filter is a data structure that can be used for checking whether an element is in a set when the set is too large to use conventional hashing techniques. It is probabilistic in the sense that false positives are possible, i.e. it might say that an element is in the set when it is really not. There are, however, never false negatives. You can not remove elements from a basic bloom filter so it is well-suited to data streaming problems where the number of elements only increases.

An example use case would be a web crawler that needs to determine whether is has already visited a URL. A small percentage of false positives will be acceptable because no significant portion of the web has only one link (read one incorrect bloom filter lookup) pointing to it.

The bloom filter is implemented as a large array of bits and several independent hash functions. Initially all bits are set to 0. An element to be added has all the hash functions applied to it. The bit corresponding to the result of each hash function is set to 1. To check whether an element has been seen before the same hashing process is used on the element. If all resulting bit positions are 1 then it has probably been seen before.

If m is the number of bits in the array and k is the number of hash functions and n is the number of elements, then the probability of a false positive is:

As an example, consider the case where m = 1 billion, n = 1 million, and k = 5. The probability of a false positive is about 0.94%. Note how each element requires more than one bit. m has to be bigger than n or else all bits would eventually be set to 1 and every lookup would be true, so the bloom filter only makes sense where the number of possible elements is large and a traditional bit array would not work.

Optimal number of hash functions also shows how to calculate m and k when n and the desired false positive rate are known.

UPDATE:

In an Ethereum JavaScript API dependency I found an implementation and they have a real life usage example.

tf-idf

Term frequency-inverse document frequency (tf-idf) is a measure of word importance in a document within a corpus. Words with the highest tf-idf often best characterize the topic of the document.

Words appearing most frequently in the corpus are not the most important words as might be expected. They are common words like stop-words. Rare words are actually the best indicators of importance, especially if they appear multiple times.

tf-idf can be represented by the following equation (thanks Online LaTeX Equation Editor):

Term frequency is calculated as the frequency of word i in document j is divided (normalized) by the maximum frequency of any word k in document j. Inverse document frequency, which accounts for words that are just more common, is calculated as the number of documents N divided by the number of those documents n the word i appears in, and then scaling that by taking the logarithm (the base of the log function does not matter).

This topic has come up in a couple of Coursera classes I have looked at--Web Intelligence and Big Data and Mining Massive Datasets--in the context of a search engine. Basically, you view each document and query (short document) as a vector of tf-idf scores, then you can find the most similar ones using cosine similarity as a way to rank the search results. Inverted indexes allow us to pre-compute much of the tf-idf score.

UPDATE:

scikit-learn has an tf-idf usage example at Clustering text documents using k-means.