MapReduce vs SQL Databases

A Comparison of Approaches to Large-Scale Data Analysis: MapReduce vs. DBMS Benchmarks

...we present the results of running the benchmark on a 100-node cluster to execute each task. We tested the publicly available open-source version of MapReduce, Hadoop [1], against two parallel SQL DBMSs, Vertica [3] and a second system from a major relational vendor.

First, as we demonstrate in Section 4, at 100 nodes the two parallel DBMSs range from a factor of 3.1 to 6.5 faster than MapReduce on a variety of analytic tasks. While MR may indeed be capable of scaling up to 1000s of nodes, the superior efficiency of modern DBMSs alleviates the need to use such massive hardware on datasets in the range of 1–2PB (1000 nodes with 2TB of disk/node has a total disk capacity of 2PB). For example, eBay’s Teradata configuration uses just 72 nodes (two quad-core CPUs, 32GB RAM, 104 300GB disks per node) to manage approximately 2.4PB of relational data. As another example, Fox Interactive Media’s warehouse is implemented using a 40-node Greenplum DBMS. Each node is a Sun X4500 machine with two dual-core CPUs, 48 500GB disks, and 16 GB RAM (1PB total disk space) [7]. Since few data sets in the world even approach a petabyte in size, it is not at all clear how many MR users really need 1,000 nodes.

In section 3.1 there's some points made about the advantages of databases over MR in relation to data integrity, "...a MR framework and its underlying distributed storage system has no knowledge of these rules, and thus allows input data to be easily corrupted with bad data. By again separating such constraints from the application and enforcing them automatically by the run time system, as is done by all SQL DBMSs, the integrity of the data is enforced without additional work on the programmer’s behalf."

They mention that "all DBMSs require that data conform to a well-defined schema, whereas MR permits data to be in any arbitrary format. Other differences also include how each system provides indexing and compression optimizations, programming models, the way in which data is distributed, and query execution strategies."

If you strip it away they are talking about text processing versus indexed data structures (and other parts of a DBMS).

For loading, "Without using either block-or record-level compression, Hadoop clearly outperforms both DBMS-X and Vertica since each node is simply copying each datafile from the local disk into the local HDFS instance and then distributing two replicas to other nodes in the cluster." The obvious difference to me would be that the SQL databases are creating "...a hash partitioned across all nodes on the salient attribute for that particular table, and then sorted and indexed on different attributes..."

For text processing, they note that the main problems with Hadoop are the start-up costs (10-25 seconds before all Map tasks start) and during the Reduce phase the cost of combining many small files. When you are comparing a fully indexed system versus text processing then you would expect the indexed system to be faster. Compression was also considered an advantage in the systems like Vertica's over Hadoop - where it actually reduced performance. It depends on the work being done whether the overhead of compression is worth the overhead so obviously - it's not explained why compression was a negative for Hadoop.

They also talk about the problems in setting up and configuring the parallel databases over Hadoop, which is not an insignificant difference when you are scaling to 100s and 1000s of nodes.

In the summary they talk about 25 years of database development and the advantages of B-Trees and column stores. It begs the question, then why wasn't a similar system used on the Hadoop infrastructure? MR is really more like distributed processing not an indexed, querying system.

If you took away the distributed layer what they are doing is comparing something like grep (a really bad implementation of grep) with Lucene or MySQL. Would anyone be surprised with the results then? A better comparison would've been comparing it against HBase or other distributed, indexed, data stores like Hive or Cloudbase.

Update: There's a good followup on the hadoop list by Jonathan Gray "Hadoop is not suited for random access, joins, dealing with subsets of
your data; ie. it is not a relational database! It's designed to
distribute a full scan of a large dataset, placing tasks on the same nodes
as the data its processing. The emphasis is on task scheduling, fault
tolerance, and very large datasets, low-latency has not been a priority.
There are no "indexes" to speak of, it's completely orthogonal to what it
does, so of course there is an enormous disparity in cases where that
makes sense. Yes, B-Tree indexes are a wonderful breakthrough in data
technology". He suggested Pig, Hive and Cascading would be more suitable for comparison.

CloudBurst

This is software that uses MapReduce to handle massive amounts of sequence data and reassemble it. It includes an explanation of the algorithmic side of things. I wonder how it compares with using de Bruijn graphs.

More MapReduce Groovy

A very good post on Cascading (covered previously), GOODBYE MAPREDUCE, HELLO CASCADING

Cascading’s logical model abstracts away MapReduce into a convenient tuples, pipes, and taps model. Data is represented as “Tuples”, a named list of objects. For example, I can have a tuple (”url”, “stats”), where “url” is a Hadoop “Text” object and “stats” is my own “UrlStats” complex object, containing methods for getting “numberOfHits” and “averageTimeSpent”. Tuples are kept together in “streams”, and all tuples in a stream have the exact same fields.

An operation on a stream of tuples is called a “Pipe”. There are a few kinds of pipes, each encompassing a category of transformations on a tuple stream. For instance, the “Each” pipe will apply a custom function to each individual tuple. The “GroupBy” pipe will group tuples together by a set of fields, and the “Every” pipe will apply an “aggregator function” to all tuples in a group at once.

One of the most powerful features of Cascading is the ability to fork and merge pipes together.

Once you have constructed your operations into a “pipe assembly”, you then tell Cascading how to retrieve and persist the data using an abstraction called “Tap”. “Taps” know how to convert stored data into Tuples and vice versa, and have complete control over how and where the data is stored. Cascading has a lot of built-in taps - using SequenceFiles and Text formats via HDFS are two examples. If you want to store data in your own format, you can define your own Tap. We have done this here at Rapleaf and it has worked seamlessly.

Scaling MapReduce

Before I started down using Hadoop I had to be roughly sure that it would scale. One of the pieces of evidence that convinced me was done by IBM in "Scalability of the Nutch Search Engine", they performed tests using real world systems and also modeled the architecture to verify it:

We observe...that the throughput, for a fixed data set size per back-end, increases with the number of back-ends.

We observe a generally good agreement between measurement and prediction for small number of back-end servers. The response time is essentially flat with the number of back-end servers, up to 500, 1000, and 2000 servers for data set sizes (per server) of 10 GB, 20 GB, and 40 GB respectively. The system is more scalable for larger data set
sizes because the work per back-end server per query increases.

They are able to service 1000+ nodes with a single server.

Who Killed Functional Programming?

MapReduce: A major step backwards. Two good responses, "The Great MapReduce Debate" and "Relational Database Experts Jump The MapReduce Shark":

MapReduce is not a data storage or management system — it’s an algorithmic technique for the distributed processing of large amounts of data.

MapReduce has the same relationship to RDBMSs as my motorcycle has to a snowplow — it’s a step backwards in snowplow technology if you look at it that way.

I don’t think the authors understand distributed processing and distributed file systems when they think reduce must rely on FTP. The Wikipedia article says “each [reduce] node is expected to report back periodically with completed work and status updates,” pretty much the opposite of the “pull” DeWitt and Stonebraker criticize.

It's a fairly specious argument that more features means better and the comparisons are not equivalent - it would be better compare BigTable or HBase with databases not MapReduce - and that'd be silly because these guys are column database proponents which is (roughly) the same as HBase/BigTable (they do say it lacks views but the idea used by Google is copying rather mutating). An example of using MapReduce to do indexing, is Nutch. The distributed filesystem is important and it should be clear by now, you'd hope, that storing the data is not the problem, processing it is.

These types of comparisons, "X does something Y doesn't therefore X is better", reminds me of "Who Killed the Electric Car?". One of the reasons the EV1 was said to have failed was because "lacking an engine, it saves the driver the cost of replacement parts, motor oil, filters, and spark plugs" (and more) - all those moving parts, that had to maintained were gone. How can you change gears if you don't have a gearbox?

Does that remind you of a database or other piece of software you use? Isn't the software industry just like the car industry? There appears to be market forces ensuring that things stay complicated and expensive to maintain.

Meanwhile, Google is getting on with it by providing services for terabytes of data (via Wired), "Building on the company's acquisition of the data visualization technology, Trendalyzer, from the oft-lauded, TED presenting Gapminder team, Google will also be offering algorithms for the examination and probing of the information. The new site will have YouTube-style annotating and commenting features."

Update: It's been noted that bet on cheap and rickety if it's 10 times better (also mentions hypertable) and that they're comments are not even wrong.

Update 2: Part of my rant was going to be the whole Apple designs things with the fewest stuff in them - that's what makes them the leader what they leave out. It's the same gist that I read at "Heavier than Air".

Fat Controller

When does MapReduce make sense and what architecture is appropriate? I don't really know; wheras Tom has some ideas. I like the idea of MapReduce in the Google architecture (as cloned by Hadoop). I like the use of a distributed file system to evenly smear the data across the nodes in the cluster. However, these choices don't always seem appropriate.

The easiest and most obvious example where this causes a problem is that sometimes you want to ensure a single entry or exit in your system (when to start or stop). Another obvious example is where the overhead of network latency overwhelms your ability to parallelize the processing. More than that, it seems that if you can order things globally then the processing should be more efficient but it's unclear to me where the line is between that and the more distributed MapReduce processing (and whether adding that complexity is always worth it).

Google's third lecture (slides) answers some of the basic questions such as why use a distributed file system. It also lists decisions made such as read optimization, mutation handling (serializing writes and atomic appends), no caching (no need due to large file size), fewer but larger files (64MB chunks) and how the file handling is essentially garbage collection. They have implemented appends as it was a frequent operation. This is something that Hadoop has yet to do and can be an issue, especially for databases (which includes the requirements for appends and truncates attached to that issue).

There is obviously some adaption needed to alogirthms to run in a MapReduce cluster. Lecture 5 gives some examples of MapReduce algorithms. It includes a breadth first search of a graph and PageRank. Breadth first is chosen so there doesn't need to be any backtracking. For graph searching, they suggested creating an adjacency matrix - 1 indicating a link in the matrix and 0 indicating no link. To transfer it efficiently they use a sparse matrix (where you only record the links - very much like column databases of course). MapReduce is similar, with the processing split as a single row per page. For both of these process there is a non-MapReduce component. For example, in the PageRank algorithm a process exists to determine convergence of page values.

Everything You Know is Wrong

• OWL can be, as the French would say, databasesque. Using OWL to implement database like integrity constraints.


• It's time to throw away your database (a paper describing the advantages of column databases) and replace it with something more flexible and scalable (using C-Store to store RDF). And column databases mean no NULLs. The problem isn't that your data is normalized it's that it's not normalized enough (also mentioned here).


• Following a question that I had earlier this year about where is the good code to learn from, I've read about 3/4 of Beautiful Code. And while I have a hard time finding C/C++ code beautiful it certainly covers some good examples of code and the process behind writing it (including Beautiful Tests, Beautiful Concurrency and MapReduce).


• Latency Moore's curse.


• Nova links to why what you thought about genes is wrong (I've actually heard this before but hadn't come across any articles).


• Scaling laws in Biology. Humans run on 100 watts or used to. Now we have cars etc. our power requirements are now 11,000 watts (making us the biggest creatures that have ever lived). And the magical number: 4.


• The Enemies of Reason is now on Google Video.

Scale Out or Drop Out

I recently came across this study done by IBM using Nutch/Lucene to compare scale-up (a single, shared memory, fast server) vs scale-out (a cluster). It showed that for the type of work performed scale-out systems outperformed scale-up systems in terms of price and performance. The IEEE article about Google's architecture in late 2002, notes that Google was spending $278,000 for 88 dual CPU 2GHz Xenons, 2 GB of RAM and each having a 80 GB hard drive. In the IBM paper, $200,000 gets you 112 blades, quad processor with 8GB of memory each with a 73GB drive as well as a shared storage system.

This scale-out approach was also recently mentioned by Tim O'Reilly's, MySQL: The Twelve Days of Scaleout.

Scale-up x Scale-out: A Case Study using Nutch/Lucene

The Nutch/Lucene search framework includes a parallel indexing operation written using the MapReduce programming model [2]. MapReduce provides a convenient way of addressing an important (though limited) class of real-life commercial applications by hiding parallelism and fault-tolerance issues from the programmers, letting them focus on the problem domain.

The query engine part consists of one or more front-ends, and one or more back-ends. Each back-end is associated with a segment of the complete data set...The front-end collects the response from all the back-ends to produce a single list of the top
documents (typically 10 overall best matches).

We see that the peak performance of the scale out solution (BladeCenter) is approximately 4 times better.

MapReduce by the Hour

Running Hadoop MapReduce on Amazon EC2 and Amazon S3

Apache's Hadoop project aims to solve these problems by providing a framework for running large data processing applications on clusters of commodity hardware. Combined with Amazon EC2 for running the application, and Amazon S3 for storing the data, we can run large jobs very economically. This paper describes how to use Amazon Web Services and Hadoop to run an ad hoc analysis on a large collection of web access logs that otherwise would have cost a prohibitive amount in either time or money.

It took about 5 minutes to transfer the data (which was compressed - Hadoop can read compressed input data out of the box) from S3 to to HDFS, so the whole job took less than a hour. At $0.10 per instance-hour, this works out at only $2 for the whole job, plus S3 storage and transfer costs - that's external transfer costs, because remember transfers between EC2 and S3 are free.

Scalability, Scalability, Scalability, Scalability

Dare Obasanjo has four recent postings on the recent Google Scalability Conference.

Google Scalability Conference Trip Report: MapReduce, BigTable, and Other Distributed System Abstractions for Handling Large Datasets:

The talk was about the three pillars of Google's data storage and processing platform; GFS, BigTable and MapReduce.

A developer only has to write their specific map and reduce operations for their data sets which could run as low as 25 - 50 lines of code while the MapReduce infrastructure deals with parallelizing the task and distributing it across different machines, handling machine failures and error conditions in the data, optimizations such as moving computation close to the data to reduce I/O bandwidth consumed, providing system monitoring and making the service scalable across hundreds to thousands of machines.

Currently, almost every major product at Google uses MapReduce in some way. There are 6000 MapReduce applications checked into the Google source tree with the hundreds of new applications that utilize it being written per month. To illustrate its ease of use, a graph of new MapReduce applications checked into the Google source tree over time shows that there is a spike every summer as interns show up and create a flood of new MapReduce applications that are then checked into the Google source tree.

Google Scalability Conference Trip Report: Using MapReduce on Large Geographic Datasets:

A common pattern across a lot of Google services is creating a lot of index files that point and loading them into memory to make lookups fast. This is also done by the Google Maps team which has to handle massive amounts of data (e.g. there are over a hundred million roads in North America).

Q: Where are intermediate results from map operations stored?
A: In BigTable or GFS

Google Scalability Conference Trip Report: Lessons in Building Scalable Systems:

The most important lesson the Google Talk team learned is that you have to measure the right things. Questions like "how many active users do you have" and "how many IM messages does the system carry a day" may be good for evaluating marketshare but are not good questions from an engineering perspective if one is trying to get insight into how the system is performing.

Specifically, the biggest strain on the system actually turns out to be displaying presence information.

Giving developers access to live servers (ideally public beta servers not main production servers) will encourage them to test and try out ideas quickly. It also gives them a sense of empowerement. Developers end up making their systems easier to deploy, configure, monitor, debug and maintain when they have a better idea of the end to end process.

And finally, Google Scalability Conference Trip Report: Scaling Google for Every User, which had some interesting ideas about search engine usage.

Update: links for 2007-07-06 includes links to the Google talk on MapReduce tasks on large datasets and other goodies (scalable b-trees, hashing, etc).

Update 2: Greg Linden has a much better list of videos and commentary.

MapReduce SPARQL

Compositional Evaluation of W3C SPARQL Algebra via Reduce/Map "Tested against older DAWG testsuite. Implemented using functional programming idioms: fold (reduce) / unfold (map)

Does that suggest parallelizable execution?"

Thanks to SPARQL's adoption of compositional semantics (algebra) it's possible. From what I can remember, while there is an implicit left to right evaluation of OPTIONAL, which does limit the top level execution, the partial results are parallelizable. Being able to execute order independent OPTIONAL while retaining the correct results for OPTIONAL is still an open question (although feasible I think).

Available from SVN here. More musings, Musings of a Semantic / Rich Web Architect: What's Next?.

Lucene for the Semantic Web

Hbase

Google's [WWW] Bigtable, a distributed storage system for structured data, is a very effective mechanism for storing very large amounts of data in a distributed environment.

Just as Bigtable leverages the distributed data storage provided by the [WWW] Google File System, Hbase will provide Bigtable-like capabilities on top of Hadoop.

Data is organized into tables, rows and columns, but a query language like SQL is not supported. Instead, an Iterator-like interface is available for scanning through a row range (and of course there is an ability to retrieve a column value for a specific key).

Any particular column may have multiple values for the same row key. A secondary key can be provided to select a particular value or an Iterator can be set up to scan through the key-value pairs for that column given a specific row key.

From the Hbase/HbaseArchitecture page:

HBase uses a data model very similar to that of Bigtable. Users store data rows in labelled tables. A data row has a sortable key and an arbitrary number of columns. The table is stored sparsely, so that rows in the same table can have crazily-varying columns, if the user likes.

A column name has the form ":" where and can be any string you like. A single table enforces its set of s (called "column groups"). You can only adjust this set of groups by performing administrative operations on the table. However, you can use new strings at any write without preannouncing it. HBase stores column groups physically close on disk. So the items in a given column group should have roughly the same write/read behavior.

Writes are row-locked only. You cannot lock multiple rows at once. All row-writes are atomic by default.

All updates to the database have an associated timestamp. The HBase will store a configurable number of versions of a given cell. Clients can get data by asking for the "most recent value as of a certain time". Or, clients can fetch all available versions at once.

The example tables given are very similar to untyped relations. This has only just become part of the nightly build.

Via, Data Parallel.

Advantages of Functional Programming

With functional programming seemingly getting more interest recently I was reminded of, "Can Your Programming Language Do This?" and "Functional programming in the Java language".

Joel said, "Lemme repeat that. By abstracting away the very concept of looping, you can implement looping any way you want, including implementing it in a way that scales nicely with extra hardware."

"Without understanding functional programming, you can't invent MapReduce, the algorithm that makes Google so massively scalable. The terms Map and Reduce come from Lisp and functional programming. MapReduce is, in retrospect, obvious to anyone who remembers from their 6.001-equivalent programming class that purely functional programs have no side effects and are thus trivially parallelizable."

MapReduce: Simplified Data Processing on Large Clusters, "MapReduce is a programming model and an associated implementation for processing and generating large data sets. Users specify a map function that processes a key/value pair to generate a set of intermediate key/value pairs, and a reduce function that merges all intermediate values associated with the same intermediate key. Many real world tasks are expressible in this model, as shown in the paper."

An interesting paper, especially if you were interested in creating massively scalable triple stores for example.