Marginally Interesting by Mikio L. Braun

A Trip to Silicon Valley: First impressions

2013-05-21T15:41:00+02:00

At the end of April, Leo and I went on a one week trip to the Valley. Over the years, we had built up a number of connections in the Valley and we thought that now was the time to go over there and meet face to face. We end up having 20 meetings over 6 days, which made for quite a schedule.

It’s one thing to know abstractly that the Silicon Valley is home to most computer related companies, and to drive down Highway 101 and see another well known company every 30 seconds or so. “Oh look, there’s Evernote”—“There’s Intel”—“I think I just saw Salesforce”, and so on.

The situation gets even more intense once you get to San Francisco. Particularly SoMa, an area probably 2 times 4 kilometers wide, seems to host every Internet startup you have ever heard of, and some bigger ones, too. Twitter, Trulia, Flurry, Dropbox, Zendesk, etc. are all in that area. It’s as if the whole Internet industry has their offices in Berlin Mitte in the area between Unter den Linden and Leipziger Straße.

We spent a lot of time in coffee shops for free Wifi, especially in the Paris Baguette on University Avenue in Palo Alto, a Korean coffee shop which reminded Leo of his time in Seoul. To get Wifi access, you had to check into the Paris Baguette on Facebook, something I found pretty neat. As it turned out, this wasn’t a new general feature of Facebook, but something being test-driven in a few coffee shops in the Valley first. We found a few more such examples, like being able to pay with bitcoins in the Coupa Cafe.

It sometimes felt as if the whole Valley was turned into one big sandbox to try out new business ideas and new pieces of technology. Already on our first day, when we sleepily sat in the sun trying to shake off our jetlag, we noticed that everyone seemed to be an entrepreneur. People were discussing business models, hacking on their websites, pitching, wherever we went. Later, people would complain that it’s so hard to hire anyone because everyone wants to be a founder.

As we started to talk to people, we also noticed people being quite open and supportive. It’s probably our German bias, but when you talk to people in Germany about your business, they quickly get defensive and start to question the merit of your whole approach. “Hasn’t that been done before” or “I think I still don’t understand what’s so great about that” are the kinder things friends of you would say. In contrast, people in the Valley seemed much more open as if there’s a general understanding that it doesn’t hurt to try. Even if people weren’t impressed by your approach they’d offer some piece of advice. It was also very common that people offered to connect you to other people which might be interested.

Originally, we had no meetings scheduled for Friday, the day when we were flying back to Germany in the evening, but in the end we had three meetings more or less back to back just because of these introduction. It felt as if we could have stayed for another week without getting bored. People later told us that they know of people who came over for three months and still could have gone on.

As someone said: The funny thing about the Valley is that although it’s all about the Internet and being connected online, actually meeting face to face counts so much.

Click here for the full article

Reclaim your data, own a piece of the cloud!

2013-04-05T16:45:00+02:00

Lately I’ve been discussing quite a bit with Leo about the current state of the ‘Net. Sure it’s nice to get all those services in the cloud for free, but in the end, you either have to worry about what exactly happens with your data, or what you can do against companies shutting down cloud based services like the Google Reader, leaving you with a pile of useless XML files, a bit like letting you take home the remnants of your car after compactification.

I used to say that the main problem is that the user is the product, not the customer, but this post by Derek Powazek convinced me that even paying for the service won’t ensure that you get decent support and control over your data.

In the end, the only thing that helps is to reclaim your data and the service itself. Just like your wordpress powered blog on your own root server will stay around as long as you pay the bills, both data and the software to make it come alive should run on a computer you control.

But how have ended up in a situation like this anyway? Here is my little history of networked computing.

In the mainframe era, computers where huge bulky machines, and only large institutions could afford to have one. People invented time-sharing operating systems to make those computers usable to many people concurrently, which were usually connected through dumb text-terminals. Those terminals mostly worked in a block-oriented manner, meaning that they presented you with a form which you would submit to the server to get the results, a bit like a form on a web page.

Obviously, services where hosted centrally, and you very much depended on the mainframe for storage and providing the service.

All this changed with the advent of the Internet and the home computer. Instead of a relatively small number of large mainframe computers you got a large network of small machines. Services like mail, ftp, and even http were designed in a way that they could run in a decentralized manner. In principle, anyone could hook up a computer to the network and run the services he was interested on his server.

Of course, you had to solve a number of technical problems, getting good bandwidth to your home was a problem, you had to use a dynamic DNS service to map changing dial-up IP addresses to a DNS entry, you had to know Linux or some other variant of UNIX, but it was possible.

Server virtualization made things a lot easier. People realized that most of the times, computers were sitting idle anyway, so why not combine them virtually in a server. That also made it possible to host a large number of servers in data centers, where they also had constant internet access, for relatively small amounts of money. (BTW, virtualization already existed in the mainframe era.)

Server virtualization and the resulting technology of putting lots of PC-type servers into racks (which look a lot like the mainframes of old from the outside) allowed companies to create massive server farms for their data intensive services.

It probably all started with Google search and Amazon. Google, because they needed to store an index of the whole web somewhere, Amazon, because millions of people wanted to use the website each day.

Lead by this example, other companies followed, and nowadays it’s entirely normal to rent out thousands of servers in the cloud (virtual or otherwise) and build services on that private armada of computers.

I’m not the first to point out that this is really just the same setup like the mainframe era, only with different technological means. While your computer is in principle able to store enormous amounts of data, and it can provide the same services as the machines in the cloud, it’s reduced to a screen to run some web browser.

So we went full circle from centralization to decentralization and back, gaining and losing control over our data and the services we need.

But there is a way out. Now it’s easier than ever to rent a piece of the cloud. We already spend enough dollars per month on our smartphones, and probably also for some cloud based services like cloud storage, why not spend a bit more money and also own a small machine somewhere in the cloud?

If that seems odd to you, have you ever noticed that a smartphone is already a bit like a small server in the cloud? It runs Linux (well at least some do), is always connected, comes typically with a few GB of local storage. In principle, you could install some dynamic DNS program on it to become a full Internet server.

Of course, managing virtual machines is still much too technical for the ordinary person. We would also need a new type of cloud based service which would keep only data which needs to be global in the company’s server farm while offloading user specific data to the user’s servers.

But technically, it’s all possible. And wouldn’t it be cool? ;)

Click here for the full article

Misconceptions about the CAP Theorem

2013-03-20T21:35:00+01:00

If you’ve ever listened to a NoSQL talk, you’ve probably come across the CAP theorem. The argument usually goes like this:

Traditional databases guarantee consistency.
The CAP theorem tells you that you cannot have consistency, availability, and fault-tolerance at the same time.
But we want to build scalable databases, so we forget about consistency.
Oh and by the way, who needs consistency anyway?

To be honest, to me this always looked like some poor excuse to not really discuss the design decisions of some NoSQL database. It’s probably just me, but I much prefer at least an attempt at an unbiased analysis of the pros and cons so that I can make an informed decision whether it fits my needs or not. But pulling this theorem out of the hat is like saying “we don’t even need to discuss this, because this theorem says impossible, ok!”

While searching for discussions of the CAP theorem, I found this excellent (but lengthy) article by Eric Brewer, one of the original authors of the CAP theorem: CAP Twelve Years Later: How the “Rules” Have Changed.

Here is my summary:

First of all, the interpretation that the CAP theorem says “you can only have 2 out of 3” is misleading. It’s not like the original proof discussed all possible choices and showed that you can have only 2 out of 3.

Instead, the original proof discusses the following situation: Say you have a distributed system which is in a consistent state (whatever that means), and now there is a Partition of the system, either an actual network failure, or some other way in which machines cannot talk to one another anymore.

Now consider what options you have when there is write request. You could wait for the partition to end in order to make sure that your system stays in a consistent state (thereby sacrificing Availability), or you could do the update partially (thereby sacrificing Consistency). So you can have only C or A in case of P but not both.

Note that there is really no way in which you could “choose P”, it was always about how to handle partitions (which are often not really partitions, but timeouts), and that includes how to detect partitions, how to behave when you are in a “partition state”, and how to bring the system back to a consistent state after a partition.

The article stresses that these are no binary decisions, but that there is rather a whole spectrum of possibly actions and strategies to choose from. It’s not about saying “I can’t have consistency and availability, so I’ll just forget about consistency”, it’s about saying “in case of a failure, availability is more important to me, therefore I will accept temporary inconsistencies, and implement strategies to clean up afterwards”.

When you look at it that way, you get a much clearer picture of how a database like Cassandra fits into this, and how their read repair, hinted handoff features work to regain consistency, although in a very lax (and eventual) way.

But it also becomes clear that it’s just not true that you cannot have distributed databases which are highly available and come with consistency guarantees at all. The article goes on to discuss recent research results which try to achieve exactly that, strategies to minimize the impact of a partition on availability and consistency, how to re-establish consistency after a partition (also in the broader database sense of having consistent cross-references between tables and satisfying other invariants)

So the next time someone tells you he doesn’t care about consistency because of the CAP theorem, ask him how he chooses P, and how he deals with the detection, handling, and cleanup of partitions.

Click here for the full article

Lunch talk: Google Reader

2013-03-14T15:35:00+01:00

As you’ve probably heard by now, Google is shutting down Google Reader on July 1, 2013. Reactions are mixed, ranging from people who say that RSS is dead anyway, or that they get all their info from other services like prismatic anyway.

RSS was dead years ago #getoverit
— Nick Halstead (@nik) March 14, 2013

Google Reader retiring is of no impact 2 me. I moved to @prismatic quite some time back. It's time for @prismatic 2 pull up 2 the next level
— Debasish Ghosh (@debasishg) March 14, 2013

Meanwhile web sites are posting lists of Google Reader alternatives like feedly, flipboard, or newsblur are compiled.

So obviously, thinkberg, who is a big fan of Reader, and I had a lot to discuss over lunch. So what is our take on this?

RSS is dead?

I agree that RSS is probably much too technical for the average user, but what people seem to forget about RSS is that it’s half of the open decentralized social network framework guys like App.net or diaspora are trying to build. You can host your own content, you can subscribe to content no matter where it’s hosted and assemble your own timeline. Also, RSS’s use is much more widespread than you’d probably think. All news sites, and blogs support RSS feeds. The obvious exception are the big social network sites like Google+, and Twitter (although you can export real-time searches as RSS feeds, which is pretty nice), because they want you to spend time on their site while they show you some ads.

It’s not perfect, of course, there are no comments or discussions (but tumblr is also much less interactive than facebook, for example), it’s pull, not push, but the openness and extensibility (in the sense that you could hook up your coffee-machine if you want to) is undisputable.

Google doesn’t “get” social - or end-user products

I think Google shutting down the Reader is mostly another example how Google is relentlessly redesigning their services to meet some grand plan without caring much about whether the user will be happy with the end result or not. It’s almost as if they are taking an engieering approach and have a hard time taking into account anything which is not a technical service requirement.

I guess it’s one thing to optimize Google’s infrastructure internally, where your dependencies are relatively few and clearly defined and you have other Google people at the other end of the line who know that these changes are necessary. Contrast this with a service used by millions of people, most of whom are don’t really want anything to change.

For example, I think people didn’t really care whether Picasa got integrated with Google+, or whether Google Docs became part of Google Drive. So it might have made sense for Google’s long-term goal to have a single social platform which incorporates all these services somehow, but most people are lost somewhere along the line.

It’s not as if Google isn’t trying to get social right, however. Brian Shih, who once was product manager for Google Reader, says on Quora:

Ironically, I think the reason Google always wanted to pull the Reader team off to build these other social products was that the Reader team actually understood social (and tried a lot of experiments over the years that informed the larger social features at the company). Reader’s social features also evolved very organically in response to users, instead of being designed top-down like some of Google’s other efforts.

Google+ to the rescue?

So what about the alternatives? What about Google+?

As thinkberg nicely pointed out, the difference is that Google Reader is (was) a productivity tool, while Google+ is a social network site. Google Reader is optimized to help you sift through many news items quickly, while Google+ wants you to linger (possible to show you as many ads as possible). In the compressed layout, Reader uses exactly one line per item, showing you dozens on a single screen. In constrast, in Google+, you seldom get more than three items per screen.

Of course, you can share items on Google+, but the old Reader was excellent in that it presented shared items also in the same compact view and by users, shared items are simply lost in the your timeline stream.

Finally, and I’m still amazed that this hasn’t changed ever since Google+ was launched, there is no easy bookmark solution in Google+. In Reader, you could star an item to bookmark it for later reading, but there is no way to see all the items you +1’d on Google+. The closes thing I’ve come across is to have a circle called “Bookmarked” and reshare to that circle to bookmark posts.

thinkberg said that this makes perfect sense for Google because they’re only interested in the +1s to improve their search results, not to provide value to the users. Of course, he was just joking, but there is some truth to this. After all, Google+ is a free service and to make sense to Google they need to leverage the information collected there in some other ways, for example in personalized search.

As thinkberg tweeted earlier

We will have to rethink the culture of free products (ad supported incl.). It’s most always a loss for the customer in the long run.
— Matthias L. Jugel (@thinkberg) March 14, 2013

the actual problem might also be that these are all free services. Such services are always hard because companies must find ways to exploit the data in some way to still make money, and customers or users have a hard time complaining if the service is changed in fundamental ways.

Or put differently, if I’d paid for Google Reader, I’d be pretty pissed now.

Having paying customers on the other hand makes everything so much easier. The company knows why it’s having a product, and it can also support small user bases. The problem is, especially with new technology, to get people to realize that a service is worth real money, but that is another topic.

Alternative RSS feed readers: there are none

So what about the other alternatives? Well, the problem is that Google Reader pretty much killed the market for classical RSS feed readers like bloglines. Most of the other companies mentioned in the various lists have taken a different approach, trying to create some nice looking newspaper-like experience from your feeds. Unfortunately, as I said above, the strength of Google Reader was to be able to quickly sift through huge amounts of information, and that’s something these newspaper-like apps cannot accomplish.

I personally like prismatic a lot, but that’s again something different. It’s pretty good at finding stuff which is interesting, but you cannot track hand selected news sources that way.

What I’d like to have

So just to compile this for the posterity, what are the features I’d like to see:

Track a number of RSS feeds (probably also other sources).
Choice between compact view (one line per item is enough) and expanded view.
Web view and mobile app, synchronized.
Bookmarking capabilities.
Sharing feature, but don’t just put everything in my timeline, let me see only shared items or even by specific user.
Hide stuff I’ve already seen.
I’d be ready to pay around 5$ per month for this.

Basically, someone give me the old Reader, and you’ll have me as a customer instantly.

Oh and in case you haven’t seen this yet:

Click here for the full article

More Google Big Data papers: Megastore and Spanner

2013-03-11T15:35:00+01:00

I recently reviewed a number of Google Big Data papers. Today I’ll talk a bit about Megastore and Spanner, two distributed databases I wasn’t aware of before. Both are distributed databases with interesting features like full consistency and transactions, something most NoSQL proponents will tell you is impossible to scale.

Now I should mention that I’m not a distributed systems guy by training, and the papers are sufficiently vague in the exact details of how they work (although they otherwise do an excellent job of describing all the essential components of such a complex system), but anyway, here is what I understood ;)

While these papers are probably more Big Data Base than Big Data Science, I still think these make an interesting read because they tell you a lot about what it takes to build a scalable and robust system.

Megastore: BigTable + transactions + schema

You might not have heard of Megastore, but if you’ve used any of Google’s products beside search, you will have interacted with it as things like GMail, Calendar, Picasa (now morphed into Google+), Android Market (now “Google Play”), or AppEngine run on Megastore.

Megastore is build upon BigTable, Google’s key-value store which inspired countless open-source NoSQL databases like Apache Cassandra. However, Megastore isn’t schema-free but lets you define tables like in a standard SQL database. The mapping to BigTable columns is straightforward and you can also specify to collocate dependent tables in the same BigTable for better performance.

Probably the most interesting feature is that Megastore is globally consistent (or ACID compliant, to be more exact). The authors argue that looser consistency is nice for scaling, but for the application developer, consistency is so much easier to work with. And actually, I think they’re right. We’ve all heard the “eventually consistency is enough” talk a few times and have come to believe that often, it doesn’t really matter. But the truth is, to know that the value you wrote is actually there, or to know that a group of related updates is rolled back if your program crashes, is very valuable.

Consistency through distributed commit log

So how does Megastore achieve distributed consistency (and by the way I’m using the term consistency you already see I’m not a distributed systems guy ;))?

The main idea seems to be that Megastore manages a distributed commit log. So-called write ahead logs are pretty common in databases to guard against failures. Every write action is first recorded in a log so that you can pick up after a crash and reapply the write operations in the log. In addition, the log also gives you a time ordering for the transactions.

Contrast this with the write actions in, for example, Cassandra: There, writes can be initiated by any node and make sure a certain number of replicates have acknowledged the write. The end effect is that different nodes might see different orderings for the updates, or some not at all. Cassandra has other mechanisms like read-repair to make sure all nodes eventually have the same data, but there is no guarantee.

So how does Megastore achieve a distributed commit log? The standard way for distributed commits seems to be two-phase commit, which however requires a master node. Instead, Google used the Paxos protocol, which is a basically a voting scheme to ensure that a majority of agents agree on a proposed value. Just to clarify, it is not about voting between a number of alternatives, it’s really about agreeing on a given value in a robust manner to ensure that at least half of the agents have noticed and agreed to the number.

The Paxos algorithm

This algorithm was originally published by Leslie Lamport (yes, the LaTeX Leslie Lamport) in a paper which was written as if it reported on some historical Greek community (also includes a bunch of Greek symbols), but if you’re interested, I recommend the ”Paxos made simple” paper which explains it in plain English.

So just to give you an idea how this algorithm works, the algorithm goes through numbered rounds. In each round, the number of the round is first announced to all nodes, and the nodes return a promise to forget about all previous rounds. If a promise was received from the majority of nodes, another attempt is made to get a majority to accept the value. If that works, the value is broadcast to all who are interested. It seems there are ways to make this all robust, including the election of a leader and the identification of the round number such that they are always larger than all previous rounds, just check the original paper for the details ;).

Now voting on a value sounds pretty abstract, but the trick used in Megastore is to use Paxos for reaching agreement on what will be appended to the commit log. Basically, the initiating node says “OK, I’d like to work to commit on this transaction here next”, and if all agree, it is ensured that the majority of nodes has synchronized commit logs.

Wrap-up: Read and write in Megastore

So on each transaction start, Megastore first identifies the last transaction committed and identifies a node which is up-to-date, or brings a node up-to- date. Then, Megastore reads the values of the transaction’s timestamp (BigTable stores multiple versions with timestamp of each value). To commit the transaction, Paxos is used to get agreement on appending the transaction to the commit log.

These are the main ideas, there is a host of other features, like “dirty reads”, “entity groups” which partition the data (and also restrict the consistency to within entity groups), a messaging system, two-phase commit for transactions across entity groups, optimizations over plain Paxos to get fewer rounds of communications, management servers to track which nodes are up-to-date, and so on.

Spanner: one atomic clock per data center

I’ll just briefly mention Spanner, yet another distributed database. The main improvements over Megastore are a SQL-like query language, better write performance (Megastore writes seem to usually take a few hundred milliseconds), and global transactions and consistency (not just within entitiy groups as in Megastore). So basically, Spanner is a distributed database which doesn’t feel different from a “normal” SQL database. Spanner has been developed to be the backend for F1, yet another RDBMS behind Google’s online ad business, and it has replaced a farm of sharded MySQL servers.

Spanner again heavily uses Paxos for various forms of coordination, but also classical two-phase commits. The main feature seems to be that Spanner is able to assign globally meaningful timestamps to commits through the use of GPS and atomic clocks. Yes, I’m not kidding, apparently, you can have rack mount atomic clocks for a few thousand dollars.

Using such clocks, Spanner globally agrees on a commit timestamp using Paxos, giving you global consistency: On a read, you get the timestamp of the last commit and retrieve the date at that timestamp. That way, Spanner also realises lock-free read-only transactions.

Distributed Consistency is hard, but not impossible

So what can we learn from these examples? What I sort of admire is Google’s courage to take on such complex problems. You often hear that global consistency is impossible to scale or that distributed transactions are a nightmare. While this is certainly massively complex technology, it is possible nevertheless. You might want to invest in some atomic clocks, though ;)

I’ve been bashing Cassandra a bit, but one also has to keep in mind that global consistency comes at a cost. Megastore writes are slow. It’s ok if you’re in an interactive application and only need to perform one transaction to complete an action, but it’s probably the wrong backend for data crunching. Likewise, Spanner’s latency is reported to be 8ms for reads on average, and 72-100ms for commits on average, which is incredibly fast given what they accomplish, but still slower than the performance you get out of Cassandra.

So it might be hard, but it’s possible. All too often, people tell us it’s impossible but sometimes they’re just defending their favorite tools feature set.

Related posts:

Big Data beyond MapReduce: Google’s Big Data papers

Click here for the full article

Stream Processing has no Query Layer

2013-03-01T16:47:00+01:00

When it comes to real-time big data, stream processing frameworks are an interesting alternative to MapReduce. Instead of storing and crunching data in batches, they process the data as it comes along, which immediately makes much more sense if you’re dealing with event streams. Frameworks like Twitter’s Storm and Yahoo’s S4 allow you to scale such computations. Similar to MapReduce jobs, you specify small worker threads which are then deployed and monitored automatically to provide robust scalability.

So at first you may think “stream processing is basically MapReduce for events”, but there is an important and significant difference: There is no query layer in stream processing (well at least, there isn’t in Storm and S4).

With query layer, I mean the capability to query the results of your computations. For stream processing, in particular, this means while the computations are still running, because you are typically consuming a never-ending stream of new events.

For example, if we consider the ubiquituous word count example, where you pipe in some constant stream of sentences (let’s say, tweets), how can you query the counts for a given word at a given time?

The answer is a bit surprising to most people I’ve talked to: There is no way you can query the result (at least from within the stream processing framework). The information is there, distributed over numerous worker thread who all see and process a part of the stream, but there is no way to access that information. Instead, results have to be periodically output, either to screen or to some persistent storage.

Now these are only toy examples, of course, but it also means that for real-world applications, you need some database backend to store your results. Depending on the amount of data you process and the level of aggregation you do (or don’t do), this also means that your stock MySQL won’t suffice to keep up with your stream processing cluster.

The same can be said of MapReduce jobs which run periodically to update some statistics, but the difference is that MapReduce doubles as the storage solution while you need an additional backend for stream processing.

So I think stream processing is good when:

you have a high frequency event stream,
have to do quite complex analyses on each event independently,
do a lot of aggregation so that there is a huge reduction in data volume.

But it’s not generally applicable when:

you need to do a lot of persistent updates which each event,
need to query results while the analysis is still ongoing.

Let me know if I’m wrong. I’d be interested in learning about some real-world experiences with scaling stream processing!

Click here for the full article

Big Data beyond MapReduce: Google's Big Data papers

2013-02-22T16:45:00+01:00

Mainstream Big Data is all about MapReduce, but when looking at real-time data, limitations of that approach are starting to show. In this post, I’ll review Google’s most important Big Data publications and discuss where they are (as far as they’ve disclosed).

MapReduce, Google File System and Bigtable: the mother of all big data algorithms

Chronologically the first paper is on the Google File System from 2003, which is a distributed file system. Basically, files are split into chunks which are stored in a redundant fashion on a cluster of commodity machines (Every article about Google has to include the term “commodity machines”!)

Next up is the MapReduce paper from 2004. MapReduce has become synonymous with Big Data. Legend has it that Google used it to compute their search indices. I imagine it worked like this: They have all the crawled web pages sitting on their cluster and every day or so they ran MapReduce to recompute everything.

Next up is the Bigtable paper from 2006 which has become the inspiration for countless NoSQL databases like Cassandra, HBase, and others. About half of the architecture of Cassandra is modeled after BigTable, including the data model, SSTables, and write-ahead-logs (the other half being Amazon’s Dynamo database for the peer-to-peer clustering model).

Percolator: Handling individual updates

Google didn’t stop with MapReduce. In fact, with the exponential growth of the Internet, it became impractical to recompute the whole search index from scratch. Instead, Google developed a more incremental system, which still allowed for distributed computing.

Now here is where it’s getting interesting, in particular compared to what common messages from mainstream Big Data are. For example, they have reintroduced transactions, something NoSQL still tells you that you don’t need or cannot have if you want to have scalability.

In the Percolator paper from 2010, they describe how the Google is keeping its web search index up to date. Percolator is built on existing technologies like Bigtable, but adds transactions and locks on rows and tables, as well as notifications for change in the tables. These notifications are then used to trigger the different stages in a computation. This way, the individual updates can “percolate” through the database.

This approach is reminiscent of stream processing frameworks (SPFs) like Twitter’s Storm, or Yahoo’s S4, but with an underlying data base. SPFs usually use message passing and no shared data. This makes it easier to reason about what is happening, but also has the problem that there is no way to access the result of the computation unless you manually store it somewhere in the end.

Pregel: Scalable graph computing

Eventually, Google also had to start mining graph data like the social graph in an online social network, so they developed Pregel, published in 2010.

The underlying computational model is much more complex than in MapReduce: Basically, you have worker threads for each node which are run in parallel iteratively. In each so-called superstep, the worker threads can read messages in the node’s inbox, send messages to other nodes, set and read values associated with nodes or edges, or vote to halt. Computations are run till all nodes have voted to halt. In addition, there are also Aggregators and Combiners which compute global statistics.

The paper shows how to implement a number of algorithms like Google’s PageRank, shortest path, or bipartite matching. My personal feeling is that Pregel requires even more rethinking on the side of the implementor than MapReduce or SPFs.

Dremel: Online visualizations

Finally, in another paper from 2010, Google describes Dremel, which is an interactive database with an SQL-like language for structured data. So instead of tables with fixed fields like in an SQL database, each row is something like a JSON object (to be more exact, the data is modeled by Google’s protocol buffer format which imposes restrictions on what fields are allowed). Internally, data is stored in a special format which makes sweeps through the data very efficient. Queries are pushed down to servers and then aggregated on their way back up and use some clever data format for maximum performance.

Big Data beyond MapReduce

Google didn’t stop with MapReduce, but they developed other approaches for applications where MapReduce wasn’t a good fit, and I think this is an important message for the whole Big Data landscape. You cannot solve everything with MapReduce. You can make it faster by getting rid of the disks and moving all the data to in-memory, but there are tasks whose inherent structure makes it hard for MapReduce to scale.

Open source projects have picked up on the more recent ideas and papers by Google. For example, Apache Drill is reimplementing the Dremel framework, while projects like Apache Giraph and Stanford’s GPS are inspired by Pregel.

There are still other approaches as well. I’m personally a big fan of stream mining (not to be confused with stream processing) which aims to process event streams with bounded computational resources by resorting to approximation algorithms. Noel Welsh has some interesting slide’s on the topic.

Related posts:

Edits: Feb 25, 2013 Clarified the Text on Dremel a bit based on Dmitriy Belenko’s comments, corrected post time.

Mar 12, 2013 Added link to follow-up post.

Click here for the full article

Yet another Big Data whitepaper

2013-01-17T15:45:00+01:00

I recently read the white paper “Challenges and Opportunities with Big Data” published by the Computing Community Consortium of the CRA. It was an interesting read, but I found it a bit too database centric, which is probably no suprise given its set of authors.

White papers are an interesting class of documents. They’re usually written from a high-level point of view and are often used to represent opinions. I often wonder what their purpose is. At least one is to form a citable piece of authority which others can then use to support whatever agenda they’re up to. For example, McKinsey’s Big Data study is continually cited to say that Big Data will have a huge economic impact in the next years.

The CCC report of course also believes that Big Data is The Next Big Thing, and discusses what it believes are the upcoming challenges. It first defines a Big Data Analysis Pipeline consisting of the steps

Acquisition/Recording,
Extraction/Cleaning/Annotation,
Integration/Aggregation/Representation,
Analysis/Modeling, and
Interpretation,

as well as overall properties like Heterogeneity, Scale, Timeliness, Privacy, and Human Collaboration (i.e. cloud sourcing). The report then discusses each of these aspects by giving a brief overview of the state of the art and then discussing future and current challenges.

It becomes pretty clear quickly that most of the authors come from a data base background, because the report outlines an approach similar to what has worked so well in the data base world. Basically, Big Data will be solved by the following two steps:

Transform raw data into a format which is more suitable for automated analysis, also dealing with noisy or incomplete data.
Separate the “what” from the “how”, just as a SQL query declaratively says what to compute, leaving it up to the database to break this down into an optimal set of primitive search operations, which we would then know how to scale.

I have my doubts concerning both steps. Don’t get me wrong, having a declarative query language which automatically scales to petabytes of data would be really cool, as well as automated ways to represent data effectively.

But from an ML point of view, data is always noisy and also highly unstructured, in particular in a way we don’t yet know how to interpret automatically. For example, if you take text, it’s pretty clear for us as humans that there is certain information encoded in the text. We have taught computers to understand some of that information using representations like n-grams, some robust form of parsing, and so on, but it is also pretty clear that there is a lot of information we just don’t know how to preserve. In other word, we just don’t know how to preserve the information contained in the data for many media types, and existing techniques will likely destroy a substantial part of that information.

Concerning a declarative query language, I think data base people probably don’t have a good idea of just how divers Machine Learning methods can be. They are inspired by biology, physics, statistics, usually use vectorial representations and linear algebra for computations, or run complex Monte Carlo simulation, all of which cannot be naturally expressed using JOINs and similar things in a relational data base system. So when the report complains that

Today’s analysts are impeded by a tedious process of exporting data from the database, performing a non-SQL process and bringing the data back.

the main reason is that the non-SQL process is so different from a relational world view that this is actually the most effective way to do the computation.

Moreover, ML algorithms are often iterative and have a lot of state, meaning that you don’t just run a query of sorts, but rather complex computations which update a lot of state again and again until your results have converged.

It’s probably possible to build such DSLs for subgroups of algorithms (for example, Bayesian algorithms, distributed linear algebra libraries, etc.), I’m unsure whether there is a general query language which is not just a generic cluster job system. Existing approaches like map reduce or stream processing come with very little assumptions on your data and mostly manage just the parallelization aspect of your computations.

Still, an interesting paper from an infrastructure point of view, but I feel like they’re missing the point when it comes to the data analysis part.

Click here for the full article

What is streamdrill's trick?

2013-01-15T16:46:00+01:00

In the previous posts I talked about what streamdrill is good for and how it compares to other Big Data approaches to real-time event processing. Streamdrill solves the top-k problem which consists in aggregating activities in an event stream over different timescales and identifying the most active types of events in real-time.

So, how does streamdrill manage to deal with such large data volumes on a single node?

Streamdrills efficiency is based on two algorithmic choices:

It uses exponential decay for aggregation.
It bounds its resource usage by selectively discarding inactive entries.

Let’s take these two things one at a time.

Exponential Decay vs. Exact Time Window

Usually, when you do trending, you aggregate over some fixed time window. The easiest way to do this is to keep counting and reset the counters at a fixed point in time. For example, you could compile daily stats by taking the values at midnight and resetting the counters.

The problem with that approach is that you have to wait for the current interval to end before you have some numbers, which is hardly real-time. You could take the current value of the counter nevertheless and then try to extrapolate in some way, which is also hard if the rate at which events come in varies a lot.

Another approach is to do a “rolling window”, for example, having the counts between now and the the same time yesterday. The problem with that approach is that you need to keep all the data somewhere so that you can subtract an event when it falls out of the window. If you want to aggregate over longer time intervals, say a month, this gets hard, as you have to potentially store millions of events somewhere.

Another approach is to use some form of decay. The idea here is that the “value” of an event decays continually over time until it is eventually zero. That way, you again get an aggregate over some amount of time, although it’s not exactly the same thing as a fixed time window. In the end, it doesn’t really matter if you understand what it is exactly you’re measuring.

The good thing is that decaying counters can often be implemented in a way which does not require us to keep all of the information, but only a few numbers per entry.

Streamdrill does exactly this, using exponential decay. This means that after a specified amount of time, the value of an event has continually reduced to one half, and then again to a quarter, and so on. We chose exponential decay because it has the nice property that a counter always decays to half its original value after a given amount of time, irrespective of the initial value of the counter. That is not the case, for example, for linear decay, where a counter starting at 1000 would take twice as long as one starting at 500.

Keeping resource usage bounded

As I said above, streamdrill throws data away. More specifically, it removes the most inactive entries to make room for new ones. The main purpose of this technique is to ensure that the resource usages of streamdrill are bounded which is important to keep streamdrill’s performance constant.

This is also something we’ve found people having a hard time to digest. After all, isn’t the whole purpose of Big Data to never throw away data?

First of all, note that there really is no way around throwing data away if you want to have a system which runs in a stable manner. Otherwise, data will just accumulate and your system is bound to become slower, unless you are able to grow it. But that costs money and adds a layer of complexity to your system you should not underestimate.

Throwing data for inactive entries away also has the nice effect that you focus on the data which is much more likely to make the top-k entries.

Second of all, as I’ve also discussed in another post, for many applications, in particular if you’re really interested in finding the most active elements, it’s completely ok if you get only approximate results. The counters you get will be approximate, but the overall ranking of the elements will be correct with high probability.

In fact, approximative algorithms are nothing to be afraid of. Approximation algorithms have been around for a long time. For many hard optimization problems, approximative algorithms are the only way we know how to tackle those problems. Such algorithms trade exactness for resource usage, but come with performance guarantees, meaning that if you can afford the computation time or memory usage, you can get the approximation error as small as you wish. The same is true for streamdrill. If you have enough memory to keep all the events, you will get the correct answers.

Streamdrill is even attractive in use cases where you need exact results (for example, in billing), because you can combine it with an (supposedly already existing) batch system, to get real-time analytics without having to invest heavily to scale your batch system to real-time.

So this concludes the mini-series for streamdrill. If you’re interested, head over to streamdrill and download the demo, or contact Leo or me on Twitter, or post your comments and questions below.

We’re currently planning what features to add for the 1.0 version and thinking the details of the licensing model. Currently we’re thinking about both standalone licenses, as well as SaaS-type offerings.

Click here for the full article

Streamdrill compared to other approaches for the Top-K-Problem

2013-01-10T23:11:00+01:00

In my last post I’ve discussed what streamdrill does: It solves the top-k problem in real-time which consists in counting activities of different event times over a certain time interval.

So far, so good, but you might wonder why you couldn’t just do this by yourself. After all, it’s just counting, right? Actually, the problem that streamdrill solves is simple, as long as:

the data volume is small
the time windows are small compared to the data volume
the number different kinds of events is small and bounded
you don’t need the results in real-time

But as soon as you have millions of events per day, wish to aggregate over days, weeks, or even months, and potentially have several million of different types of events, the problem gets quite complicated.

You may end up in such a situation faster than you think:

Event rates may rise up beyond what you originally envisioned (in particular if you’re successful. ;-))
The number of different event types may explode. This might either be because the underlying sets are already large (say, IP addresses, or users in a social network), or because you are tracking combinations such that the sizes multiply.

Let’s dig deeper into this problem to better understand why this problem quickly gets hard.

One thing you need to keep in mind is that the other solutions we will discuss still involve some amount of coding. So if you compare streamdrill against Hadoop, you would need to do a non-trivial amount of coding for Hadoop, because Hadoop is a general purpose framework taking care of the scaling, but doesn’t solve the top-k problem out of the box.

	streamdrill	Standard SQL	Counters	Stream Processing
solves top-k out of the box	✓	✕	✕	✕
real-time	✓	✕	(✓ \$\$\$)	✓
focusses computation on "hot set"	✓	✕	✕	✕
memory based for high throughput	✓	✕	✕	✓
persistent	(✓ ⌚)	✓	✓	(✓)
scales to cluster	(✓ ⌚)	✕	✓	✓
exact results	(✓ ⌚)	✓	✓	✓

✓ = yes, ✕ = no, (✓) = possible, (✓ \$\$\$) = possible, but expensive, (✓ ⌚) = possible, but not yet.

Approach 1: Store and crunch later

Let’s start by using a traditional approach based on some SQL database. To do that, you would create a table with a timestamp column and columns to reflect the fields in your data. Then you would just pipe in each event into the database. To get the counts you would run something like SELECT count(*) FROM events WHERE timestamp > '2012-12-01 00:00' AND timestamp < '2012-12-31 23:59' potentially also grouping to focus on certain types of events, and adding an ORDER BY count(*) clause to get the most active elements.

There a number of problems with this approach:

If you use a normal disk based database, you will be able to add only a few hundred, at best thousands, events per second,
As you add more data, your database will become slower over time (and you will be adding a lot of data!). Leo has a number of nice benchmarks on his blog for MongoDB and Cassandra insertion performance.
Just adding the data is not enough, you also need to crunch the whole data to compute the activities. But the longer the time window, the longer will the query take to run.
While the query runs, there will be considerable load on your server, making the addition of events even slower.
Eventually, you will get so slow that your results will already be a few minutes or even hours old once you get them. Hardly real-time.

What’s more, you’re probably only interested in the top 100 active elements, so most of the computation is spent on data you’re not interested in.

In other words, putting the data into some big database and crunching it later won’t really scale. If you’ve got a lot of money on the side, you can employ some form of clustering using map reduce or a similar approach to bring down the times, but the bottom line is the same: You crunch a lot of data which you don’t really need, and the problem will only become harder if you get more and more data. And “harder” also means a lot of coding and operations work (which you don’t have if you just use streamdrill ;)).

Approach 2: Just keeping the counters

So just storing the data away and crunching it later won’t work. So how about doing the counting on the spot? That way, you wouldn’t have to store all those duplicate events. Note that just keeping the counters isn’t sufficient, you also need to maintain the global index such that you can quickly identify the top-k entries.

Using a complex event processing framework like Esper would also be a way to reduce the coding load, as Esper comes with a nice query language which let’s you formulate averages over time windows in compact way.

Let’s assume your data doesn’t fit into memory anymore (otherwise it won’t be Big Data, right?). One option is to again store the counters in a database. However, just as in the previous example this restricts the number of updates you can handle. Also, you will generate a lot of changes on the database and not all databases handle that amount of write throughput gracefully. For example, Cassandra only marks old entries for deletion and cleans up during the compaction phases. In our experience, such compactions will eventually take hours and put significant load on your system, cutting the throughput in half.

And again, most of the time is spent on elements which will never make the top-k entries.

Approach 3: Stream processing frameworks

Instead of keeping counters in a database, you could also try and scale out using a stream processing framework like Twitter’s Storm, or Yahoo’s S4. Such systems let you define the computation tasks in the form of small worker threads which are then distributed over a cluster automatically by the framework, also keeping the counters in memory.

While this looks appealing (and in fact, allows you to scale to several hundred thousand events per second), note that this only solves the counting part, but not the global index of all activities. Computing that in a way which scales is non-trivial. You can collect the counter updates at a worker thread which then maintains the index, but what if it doesn’t fit into memory? You could partition the index, but then you’d have to aggregate the data to compute queries, and you’d have to do this yourself, so again, a lot of complexity. The above stream processing frameworks also don’t come with easy support for query, so you’d need to build some infrastructure to collect the results yourself.

And again, you also have a lot of computation for elements which will never show up in your top 100.

In summary

While conceptually simple (in the end, you just count, right), the top-k problem which streamdrill addresses becomes hard if there are more things to count than fit into memory, and the event rate is higher than what you can write to disk.

Finally, let’s discuss some of the differences:

As streamdrill is memory based, all the data is lost when streamdrill crashes or is restarted. However, we already have functionality in the engine to write snapshots of all the data, but those aren’t available yet via the API streamdrill.
Right now, streamdrill does not support clustering. We just haven’t found it necessary so far, but it’s something that is possible and will be included soon.
Finally, as I’m going to explain in more depth in the next post, streamdrill is based on approximate algorithms which trade exactness versus performance. Again, if exactness is really an issue, you can get it by combining with one of the other technologies. This is possible, but not our top priority for now.

In the next post, I’ll explain how streamdrill approaches the problem.

Related posts:

Click here for the full article