Continuous Analytics on Graph Data Streams Using WSO2 CEP

Is iPaaS solving the right problems? Not knowing the fundamental difference between iPaaS and dPaaS could cost you down the road. Brought to you in partnership with Liaison Technologies.

ACM Distributed and Event-based Systems (DEBS) Grand Challenge is a yearly competition where the participants develop an event-based solution to solve a real world high-volume streaming problem. This year’s challenge focuses on analyzing the properties of a time evolving social-network. The data for this year’s event has been generated using Linked Data Benchmark Council (LDBC) social network data generator.

The ranking of the solutions is carried out by measuring their performance using two performance metrics: (1) throughput and (2) average latency.

WSO2’s been submitting solutions to the grand challenge since 2013, and our previous grand challenge solutions have been ranked as one of the top solutions among the submissions. This year, too, we submitted a solution using WSO2 CEP/Siddhi. Based on its performance, this year’s solution has also been selected as of the top solutions and we were invited to submit a full paper to the DEBS 2016 conference. This paper was presented in the DEBS conference on 22 June 2016.

In this blog I’ll present some details of our solution which is based on WSO2 CEP.

This year’s challenge involves analyzing a dynamic (evolving) social-network graph. Specifically, the 2016 Grand Challenge targets following problems: (1) identification of the posts that currently trigger the most activity in the social network, and (2) identification of large communities that are currently involved in a topic. A brief description of the two queries is given below (more details about the two queries can be found here.)

The goal is to find k comments that were created not more than d seconds ago which have the largest range where the range of a comment is defined as the size of the largest connected component in the graph defined by persons who have liked that comment and know each other.

The following figures illustrate the architecture of the two queries. Note that in our implementation, we execute the two queries in parallel (the aim is to utilize all the machine cores)

Let me now provide some details about the architecture of the two queries. For each query there is a dedicated data-loader thread (T1, T2, T3, T4, T5) for reading each data stream. The data (i.e. tuples) read by a data loader is placed in a blocking queue. The blocking queue provides a place for the data loader to store events without having to wait for the event-ordering threads to consume them. The loader threads also keep the event-ordering threads busy by providing a backlog of events to consume (note: the maximum capacity for a blocking queue can be specified when the queue is constructed. This parameter can be used to address the excessive queue expansion problem which may occur when the production rate is higher than the consumption rate.).

There are two dedicated event-ordering threads for ordering query 1 (T6) and query 2 (T7) streams. These threads fetch the events from queues and order them based on their event time-stamps. Note that events in queues are already ordered based on their time-stamp (the time-stamp here refers to logical time). The main functionality of the event-ordering thread is to order the steam which is placed in the ring buffer. For example, in the case of query 1, it orders comment and post streams and place these invent in the ring butter. The event processing threads (Q1 and Q2) fetch the events from the ring buffer and process them according to the query logic. Note that event-ordering threads act as producer threads while the processing threads represent consumers.

As illustrated below the query 1 data structure consists of three maps (post-map,post-score map and comment-post map) and ten time windows.

When a new post arrives, it is registered in the post and post-score maps.