Druid and Pinot low latency OLAP design ideas

0. Introduction

Imagine a data schema that can be described as (M, D1, .. , Dn, {(t,v)} ) where every entry has

• M -  a measure (e.g. Site Visits)
• Di - a few dimensions (e.g. Country, Traffic Source)
• {(t,v)} - a time series with a measure value for each timestamp

Calling it a time series OLAP could be a misnomer. Time is just another dimension in a typical OLAP schema. Time series usually implies just a metric without dimensions (e.g. server CPU usage as a function of time). Nevertheless in real life systems dealing with such data is quite common. They also face similar challenges. Chief among them is the need for low latency/interactive queries. So traditional Hadoop scalability is not enough. There are a couple systems out there that are less well known than Spark or Flink but seem to be good at solving the time series OLAP puzzle.

1. Druid ideas 

• For event streams, pre-aggregate small batches (at some minimum granularity e.g. 1 min)
• Time is a dimension to be treated differently because all the queries use it
• Partition on time into batches (+ versions), each to become a file
• Each batch to have (IR-style): a dictionary of terms, a compressed bitmap for each term (and so query filters are compiled into efficient binary AND/OR operations on large bitmaps)
• Column-oriented format for partitions
• A separate service to decide which partitions to be stored /cached by which historic data server
• Metadata is stored as BLOBs in a relational DB
• ZK for service discovery and segment mappings
• Historical and real-time nodes, queried by coordinators separately to produce final results

2. Pinot ideas

• Kafka and Hadoop M/R based
• Historical and real-time nodes, queried by brokers
• Data segment : fixed-size blocks of records stored as a single file
• Data segments
• Time is special. Ex: brokers know how to query realtime and historic nodes for the same query by sending queries with different time filters. Realtime is slower because it aggregates at different granularity ("min/hour range instead of day")
• ZK for service discovery
• Realtime has segments in-memory, flushes to disk periodically, queries both.
• Multiple segments (with the same schema) constitute a table
• Column-oriented format for partition for segments
• A few indices for a segment (forward, single-value sorted, )

3. Common threads

• Column oriented storage formats
• Partition on time
• Information Retrieval/Search Engine techniques for data storage (inverted index)
• ZK is a popular choice of discovery service

4. In other news

There also smaller scale products using search engine technology for OLAP data storage.

Networking frameworks revisited

Talking to a former colleague the other day I learned about recent changes in the world of NIO frameworks. The guy who led the MINA project joined JBoss and designed a new framework known as Netty. Previously, JBoss Remoting 3 was supposed to be based on MINA2.

Now the original MINA and Netty designer is on Remoting team as well so I feel I cannot make sense of all this proliferation anymore (even not taking into account Grizzly). It remains to be seen if some grand unification/survival of the fittest will ensue.

While looking at JBoss documentation I also found that they have a new NIO-related project. It makes me wonder how dead Java is if a single company can produce so much new stuff in a pretty obscure niche.

Event Stream Processing

I have just stumbled upon this new approach to event-based design. A wikipedia example says it all in 7 lines. The idea is to use an SQL-like language to create filters (capable of event correlation, typically limited to a certain period) and run a continuous stream of events through them. There is even an open source implementation. At first glance most currently available products are used in algorithmic trading (whatever was left of it after the crisis :) ).

To me this concept seems to be a kind of rule engine and so potentially there are multiple domains where it might be applied. On the other hand it makes me wonder whether it is indeed a re-packaged rule engine and in such a case what is so different about it.

Performance is the name of the game because ESP is positioned as a soft real-time technology. It must be pretty interesting to read how they manage to achieve their declared ability to process a huge number of events per second on commodity hardware.

Terracotta at JavaOne

Today I went to see what was going on at JavaOne which is held a couple of blocks from my office. The Pavilion was not much different from last year so I ended up talking mostly to Terracotta guys and brainy Taylor in particular. I was asking different nasty questions about their approach and he was kind enough to talk to me for at least half an hour. Right after he talked to Brian Goetz himself.

My opening salvo quite naturally was about their messaging framework. I was essentially told that JGroups clusters of more than 4 servers had been widely known and Apache Tribes did not fit exactly and bug-fixing was insufficiently quick. From what I know the whole JBoss stack (at least in its JBoss AS 5 reincarnation) is supposed to depend on JGroups and so I am at a loss to reconcile such contradictory statements. Probably the key is the number of servers in a cluster.

At face value it means there is no open source reliable messaging framework capable of scalability beyond a couple of servers. Taking into account that pretty much anything which is not yet "in the cloud" is clustered nowadays that sounds odd. And it drives home the idea of reliable messaging as a truly challenging thing to make of production quality.

I learned more about there positioning as well. They are after the middle market of, roughly speaking, up to 50 servers in domains such as web application. Which I guess implies that JBoss is a competitor more than Coherence/Gigaspaces going after larger clusters in Finance.

An active-active L2 server configuration is expected by the end of the year although the common belief is that the 10 seconds required to switch to a backup server now are tolerable. From how I understood it they are planning to send separate updates to both L2 servers instead of multicast or replication between the two. I might have misconstrued something though.

We talked about their paradigm a little bit. I admit to being rather uncomfortable with it because they are the only company I know literally exploiting the conceptual similarity of concurrent and distributed systems (i.e. CPUs sitting on the same bus differ from servers in a cluster only by communication delays much more pronounced in the case of a LAN). It is so different from pretty much any product (exposing a real API in terms of actual interfaces in, say, JCache as opposed to delimiting transactions with monitorenter/monitorexit pairs) that either they have invented the best thing since sliced bread or they are likely to fail as mavericks. They might as well be the next "the network is the computer" after all.

The foundational paradigm of Terracotta as a distributed JVM (complete with a DGC) evokes the same kind of argument as JVM used to ten years ago. Back then the idea of Java performance comparable with C++ was ridiculous although it was said at the very beginning that JIT-style dynamic optimizations would do the trick one day. It looks like JVM guys have pulled the trick after all so this lesson may have significant implications for Terracotta.

As an example, it can detect that a particular instance is used exclusively by one L1 server and transfer lock ownership to that L1 server from the central L2 host and so effectively avoid using distributed locking. As a result we have a sort of buddy replication (between L1 and L2). Like a JVM silently eliminating synchronization in a sequential program. Theoretically neat :)

One thing I can say safely is that in contrast to many companies in this field Terracotta is not afraid to share its source code. They do not pretend like Coherences of this world do that someone can steal anything from the code base and ruin their empire (anybody heard of a new M$ after the windows source code was leaked on the net?). As a developer I believe that code quality says a lot about the corresponding system (not to mention things one can learn from a large successful system) and I applaud Terracotta for their bravery.