Graph processing for really big data

MapReduce implementations of graph algorithms like PageRank and adsorption scale to millions of nodes on a cluster of around 50 machines, but if you want to process billions (or even tens of millions, depending on your algorithm) then you need a different framework.  Google uses Pregel, about which they've said little except that it was inspired by the Bulk Synchronous Parallel model for parallel programming.

So the announcement of a BSP package for Hadoop in the Apache HAMA project could be an interesting one to watch.  There's even a BSP hello world, although getting further may be hard work with the current level of documentation.

MapReduce algorithm design

Data-Intensive Text Processing with MapReduce is a new book by Jimmy Lin and Chris Dyer of the University of Maryland.  It's due for publication later this year but a full draft is already available as a pdf.  The book shows how you can implement a variety of useful algorithms on a MapReduce cluster, including graph algorithms such as breadth first search and PageRank, and parameter estimation for latent variable models, with a detailed explanation of how to do this for Hidden Markov Models, and even a sketch for Conditional Random Fields.  Although the book is a practical manual, the algorithms are given in simple pseudo-code rather than Java classes intended for use on a Hadoop cluster.  This has huge advantages for readability, and makes it much easier for the authors to draw out some generally applicable design patterns for MapReduce algorithms.

The order inversion pattern is a nice trick that lets a reducer see intermediate results before it processes the data that generated them.  Lin and Dyer illustrate this with the example of computing relative frequencies for co-occurring word pairs e.g. what are the relative frequencies of words occurring within a small window of the word "dog"?  The mapper counts word pairs in the corpus, so its output looks like
((dog, cat), 125)
((dog, foot), 246)
...
But it also keeps a running total of all the word pairs containing "dog", outputting this as
((dog,*), 5348)
Using a suitable partitioner, so that all (dog,...) pairs get sent to the same reducer, and choosing the "*" token so that it occurs before any word in the sort order, the reducer sees the total ((dog,*), 5348) first, followed by all the other counts, and can trivially store the total and then output relative frequencies.  The benefit of the pattern is that it avoids an extra MapReduce iteration without creating any additional scalability bottleneck.

Other patterns explained in the book include the pairs and stripes approaches to produce large sparse matrix mapper output, in-mapper combining to limit the amount of mapper output written to disk (a common scalability bottleneck in MapReduce), and value-to-key conversion for relational joins of large datasets.

All in all, this book is a great complement to Tom White's Hadoop book.  An extra plus is that the pseudo code can be translated with virtually no effort into dumbo code for Hadoop.  I can see Data-Intensive Text Processing and dumbo becoming a standard way of teaching MapReduce quickly and excitingly in the classroom.

Last.fm are hiring!

If you're interested in the problems discussed here, and want to get paid for researching, designing, implementing and maintaining solutions to them, we're hiring.  See this post for a list of more great reasons to work at Last.fm.  We also sometimes take interns and like to collaborate with academic researchers.

Adsorption: scalable graph-based everything

Graph-based machine learning algorithms have a reputation for scaling badly, so I've enjoyed reading two papers describing a simple graph algorithm that does scale easily and which can be applied to a large range of problems.  Video suggestion and discovery for YouTube: taking random walks through the view graph by Shumeet Baluja and seven colleagues from Google, and Ranking and semi-supervised classification on large scale graphs using map-reduce by Delip Rao and David Yarowsky of Johns Hopkins University, both show how Label Propagation has a simple implementation in the map-reduce framework which is closely related to PageRank.

Imagine a graph in which you have labels only for some nodes, but where the edges and their associated weights show the connections between related nodes and their relative strengths.  Label Propagation allows you to infer labels for the remaining unlabelled nodes, or additional labels for those already labelled.  The Google paper considers a graph in which nodes represent videos and users, and edges between them represent views on YouTube.  User nodes are labelled with the videos they have viewed.  We can generate recommendations for a user by propagating additional labels from other users connected to them by many short paths i.e. users who have viewed some of the same videos.

The basic algorithm is so simple that, using the elegant dumbo python wrapper for Hadoop, we can write the whole thing in a few lines of code.

Let's suppose we have a log of fruit consumption here at Last.fm HQ:

norman  orange  1
norman  orange  1
mark    apple   1
klaas   orange  1
mark    banana  1
mark    apple   1
mark    apple   1
norman  banana  1
klaas   pear    1
ricky   banana  1
olivier cherry  1
norman  orange  1
klaas   cherry  1
olivier banana  1


First of all let's create a graph from this logfile.  We map the log entries:

def parse_log(value):
    user,item,count = value.split("\t")
    return user,item,int(count)
  
def map_log(key,value):
    user,item,count = parse_log(value)
    yield ((USER,user),(ITEM,item)),count
    yield ((ITEM,item),(USER,user)),count  
  
and sum the counts with a dumbo.sumreducer to create the edges.  Next we map the edges

def map_edges(key,value):
    yield key[0],(key[1],value)  
  
In the reducer we output adjacency lists, also adding a shadow or dummy node for each user, to which the observed label distributions are clamped.  We use the dumbo getpath decorator to output the adjacency lists and label distributions into different directories:

@opt("getpath", "yes")
def reduce_edges(key,values):
    values = list(values)
    yield ("adjacency_lists",key),values
    if key[0] == USER:
        yield ("adjacency_lists",(DUMMY_USER,key[1])),[(key,INJECTION_WEIGHT)]
        dist = normalise_dist(values,MAX_LABELS_PER_NODE)
        yield ("label_dists",(DUMMY_USER,key[1])),label_dist      
      
Here's the function we use to normalise and prune label distributions, and the dumbo runner for this job:

def normalise_dist(dist,max_labels):
    dist = sorted(dist,key=itemgetter(1),reverse=True)[:max_labels]
    norm = sum(weight for label,weight in dist)
    return [(label,weight/norm) for label,weight in dist]

def runner(job):
    job.additer(map_log,sumreducer)
    job.additer(map_edges,reduce_edges)

Now we're ready to propagate the label distributions.  We do this iteratively, on each iteration sending the distribution at each node to each of its neighbours.  First we use some dumbo magic to join distributions to adjacency lists.  We just need to write a reducer:

class Reducer(JoinReducer):
    def primary(self,key,values):
        self.label_dist = values.next()
    def secondary(self,key,values):
        yield key,(self.label_dist,values.next())
      
Then we transmit the distribution from each node to its neighbours:

def map_propagate(key,value):
    label_dist,adjacency_list = value
    for node,weight in adjacency_list:
        yield node, [(label,prob*weight) for label,prob in label_dist]

Finally we sum and normalise the incoming distributions at each receiving node:

def reduce_sum_normalise(key,values):
    dist = defaultdict(lambda:0)
    for d in values:
        for label,prob in d:
            dist[label] += float(prob)
    dist = normalise_dist(dist.items(),MAX_LABELS_PER_NODE)
    yield key,dist

Here's the runner for this job:

def runner(job):
    multimapper = MultiMapper()
    multimapper.add("label_dists", primary(identitymapper))
    multimapper.add("adjacency_lists", secondary(identitymapper))
    job.additer(multimapper,Reducer)
    job.additer(map_propagate,reduce_sum_normalise)

And that's it: a massively scalable graph-based recommender in under 50 lines of python.  Well ok, we still need to interpret the ouput.  We started with an observed distribution at the dummy node for Norman like this:

(3, 'norman')   [((2, 'orange'), 0.75), ((2, 'banana'), 0.25)]


and after a few iterations our recommender has inferred this at his node:

(1, 'norman')   [((2, 'orange'), 0.45735654387297453), ((2, 'banana'), 0.28646536729600702), ((2, 'cherry'), 0.11646859085648147), ((2, 'apple'), 0.074826725260416671), ((2, 'pear'), 0.064882772714120379)]


So next lunchtime I'll suggest he tries some cherries.

Integrating social relationships into a music recommender

It's always nice to see people working with Last.fm's huge quantity of publicly available data, especially when they include several researchers from IBM's famous research labs. In their ACM Recommender Systems 2009 paper Augmenting collaborative recommender by fusing explicit social relationships, Q. Yuan, S. Zhao, L. Chen, Y. Liu, S. Ding, X. Zhang and W. Zheng report some experiments incorporating Last.fm's friend and group membership relationships into more conventional Collaborative Filtering systems.

Their first recommender is a user-based CF system i.e. the score given to a new recommendation of Muse for user Jim is a weighted average over all the scores (ratings, scrobbles, etc.) for users who already listen to Muse, and the weights are similarities computed between Jim and each of Muse's fans.

Yuan et al. start with a baseline CF recommender in which the similarity between users is based on cosine distance between their vectors of artist scrobbles. For reasons that aren't obvious to me, the scrobble counts are first mapped to binary values, e.g. 1 if a user has scrobbled Muse three or more times, 0 otherwise. In their first experimental recommender, these similarities are augmented by a weighted combination of similarities based on friend and group membership relationships. The relationship similarities are also computed as cosine distances between binary vectors, so the friendship similarity between two users expresses the extent to which they have friends in common, and likewise for group membership.

The second experimental system is a graph-based recommender, built on a graph of user, artist and group nodes. The graph starts with the obvious edges between users and the artists they listen to, and the groups they belong to. Then edges are added between pairs of users whose relationship similarities are more than some threshold, so each user is connected to other users with similar friends or belonging to similar groups as well as to their 'real' friends. Like the binary CF vectors in the first system, the edge weights are all set to 1. Finally user-user similarity scores are extracted from the graph using a Laplacian matrix pseudoinverse algorithm: in practice the computation required is similar to inverting a matrix containing concatenated artist, friend and group vectors for each user.

After these nice simple suggestions for combining social relationships and CF, the experiments themselves are a bit of a disappointment, mainly because the dataset they use contains only 1000 artists and users. The evaluation metrics are precision, recall and f-value over the top few recommendations, where a recommendation is considered correct if the user actually scrobbled the reommended artist at least three times. In the CF recommender, using weights that seem to be rather highly fitted to the small dataset, there is a 2-3% improvement in f-value over the top five recommendations. The graph recommender shows an 8% improvement, but they don't give baseline results from a graph not including the social relationships, so it's not clear whether this is due to the extra social information or just the different algorithm.

In fact using such a small dataset raises all sorts of problems in interpreting the results:

• it's hard to tell if absolute values are good or bad
• is it sensible to evaluate the top-5 recommendations from a pool of only 1000 artists?
• are the improvements significant?
• are the various parameters overfitted?
• does the graph method scale?

It makes me suspect that IBM are just playing in the recommender space, while others are clearly serious about it.

Getting names right: an unsolved problem

One of the best things about working at Last.fm is the opportunity to do applied research. Not so much an opportunity, in fact, as a necessity, since many of the problems we tackle in the Music Information Retrieval team have no established solution. One challenge which I'm working on at the moment is deciding which of the millions of artist names in our catalogue may in fact refer to the same people; and likewise for albums and tracks. Where we find more than one spelling or version of a particular name we then also have to decide if one of them should be chosen as the correct one, and the others automatically corrected to it.

Historically the first of these problems has been well studied under several different names in different fields of computer science, including duplicate detection, record linkage and entity or coreference resolution. Funnily enough, in the case of a music catalogue, detecting duplicates is at least partially solved by the beauty of audio fingerprinting, which tells us with high confidence if two tracks are the same or not. The second task, usually known as name canonicalization, is much more difficult for us, and unfortunately has grabbed much less academic research attention over the years.

One recent exception is a 2009 paper on coreference resolution by Michael Wick, Aron Culotta, Khashayar Rohanimanesh and Andrew McCallum. This describes a joint graphical model (a Conditional Random Field) that can learn coreferences and canonical names at the same time. This requires reasoning about both entities and the various mentions of those entities. To do this you need to compute aggregated features that express the similarity of entire clusters of mentions to entities or other clusters of mentions. Real-valued features can be aggregated in a conventional way, for example the string distance between two groups of names can be expressed as the average or maximum of the pairwise distances. Boolean features such as exact matches can be aggregated with logical operators - a nice idea that was new to me - for example we can say that all of set of mentions of a track match a given title, or that at least one of them does, a majority of them does, etc.

According to Wick et al., doing canonicalization jointly with entity resolution leads to a big improvement in the number of duplicates identified. Canonical entities are estimated as centroids of each group of related names, where distance is measured with a tunable edit distance. This is based on the assumption that spelling variations are essentially random typos: this is frequently not true for artist names, where variations such as Avengers and The Avengers, or An Cafe and アンティック-珈琲店- are surprisingly common. Last.fm's catalogue includes huge amounts of metadata supplied by users with their scrobbles, typically in the form of artist, album and track fields in the ID3 tags of mp3 files. As a result, even by the most conservative estimate, many millions of artist names in our catalogue need correction. Trusted catalogues such as MusicBrainz contain only around 500k aritsts, making both the problem of canonicalization, and the question of how to evaluate any proposed solution, an important practical issue for us. Despite their nice results on entity resolution, Wick et al. don't evalute the results of their canonicalization itself at all, saying simply that “we have no data on which to evaluate [its] performance”. This is a pity, and avoids one of the main research challenges here, because shortage of trustworthy data is only to be expected in any scenario where large-scale automatic canonicalization is necessary.

At Last.fm we have come up with several different measures to evalute our correction system. Firstly we compute the overall likelihood of our mappings to canonical names based on probabilistic estimates of the accuracy of trusted catalogues, and of the proportion of names in our catalogue which are likely to be incorrect. This turns out to be equivalent to assigning different weights to false positive and false negative corrections, and estimating their total weight over all our corrections. Unfortunately the required probabilities or weights are very hard to estimate accurately, so this measure remains rather speculative. Secondly we use a small groundtruth marked up by hand, which works well but clearly doesn't scale. Finally we look at the votes we receive for different forms of each artist name, through the suggest a correction feature on artist pages: this gives us an approximate measure of precision, based on the number of artists where users vote strongly against our corrections to canonical names, and recall, based on the number of artists where users vote strongly to say that we need to make a correction where we aren't making one. The performance estimates produced by these different measures vary, but the precision of our system looks fairly good, at over 97% even according to the worst estimate. Our estimates of recall vary a lot, but it may be as low as 40%, leaving plenty of headroom for making more corrections.

So for any new researchers in IR looking for wide open research topic with practical applications, name canonicalization is well worth a look!