Hierarchical graphs are multigraphs, which have as vertices the leaf nodes of a tree that lays out a hierarchy, and as edges the interactions between the entities represented by these nodes. In this paper we deal with the management of records that are the edges of such a graph by describing a model that fits well in a number of applications, many of which deal with very big volumes of streaming distributed data that have to be stored in a way so as their future retrieval and analysis will be efficient. We formally define a partitioning schema that respects the hierarchy tree, and apply these ideas by using well known open source big data tools such as Apache Hadoop and HBase on a small cluster. We built a framework on which we examine some basic policies for the partitioning of such graphs and draw interesting conclusions regarding the quality of the partitions produced and their effectiveness in processing analytical queries drawn from the imposed hierarchy.
There are numerous applications such as management and
visualization of Telecommunications data [
You can see such an example in Figure 1 which presents a small part of the hierarchy of locations in Greece. In this Figure, Attiki and Messinia are states of Greece, while Athens, Piraeus and Kalamata are cities in these states. The unlabeled nodes represent This research has been co-financed by the European Union (European Social Fund ESF) and Greek national funds through the Operational Program ”Education and Lifelong Learning” of the National Strategic Reference Framework (NSRF) - Research Funding Program: RECOST
Greece Attiki
Messinia Athens
Piraeus
Kalamata subscribers of a telephony network. Then, each call serviced by the telephony network instantiates a new directed edge in the graph, between the respective vertices (caller and callee) located at the leaves of the hierarchy tree, as is shown in the Figure. This hierarchy is exploited in order to help pose queries that seek to retrieve certain records for further analysis. For instance in order to calculate statistics on the out-of-state calls originating in Athens, this intention may be described by a query edge between Athens and Greece in the tree. Similarly, query edge (Attiki, Attiki) denotes the set of calls originating and terminating within this state. The aforementioned query edges are shown in Figure 1 (dashed lines). Another example where similar hierarchical graphs exist, is social networks where users are organized in groups according to their location or other characteristics such as age or interests and we need to record the interactions between them. Users (e.g. analysts) of such data often need to answer queries regarding interactions or distributions of records between hierarchy groups not necessarily belonging to the same level of the hierarchy.
In the aforementioned applications, this kind of graphs can grow to enormous size. For instance a large telecom provider may service hundreds of millions of calls per day, each triggering a new edge in the graph. Moreover, this data is distributed by nature as it is being streamed from distant locations (e.g. call centers, web hosts, ip routers). Thus, we need solutions that can cope with the volume, but also with the streaming and distributed nature that characterize this kind of data.
In our work we address these challenges by moving the storage
and processing of these graphs to the cloud. We propose a
system that uses the distributed data store HBase [
We revisit the problem of managing massive hierarchical graphs that are streamed by many applications of interest. Our techniques utilize emerging computational and data management platforms for manipulating large, dynamic and distributed collections of records in a cluster of machines. Available parallelism is exploited via a dynamic partitioning scheme we propose for the streamed records.
We formally define the space of choices for partitioning the streamed graph, while respecting the hierarchy tree that is superimposed over its nodes. We then present a number of interesting partitioning policies, and describe the details of the system we built for implementing our framework while utilizing off-the-shelf tools.
We present an experimental evaluation of our system using a small cluster of machines. Our results demonstrate the efficiency of our system in managing massive graphs scaling to millions of edges. We also provide a comparison among the partitioning policies we implemented based on the results of a number of experiments that we conducted.
The rest of the paper is organized as follows. In Section 2 we discuss related work. In Section 3 we formally introduce our framework, discuss the type of graph data and queries we consider. Then, we describe a partitioning scheme based on the tree hierarchy that accompanies the graph data, a number of partitioning policies that we implemented and discuss the architecture of our system. Section 4 presents our experiments and Section 5 contains concluding remarks.
Interest in graphs and their applications in data management has
been renewed due to the wide spread of fields such as social
networks and the semantic web. In the same time there is a profound
need for the efficient management of big distributed data. As a
result we can see a lot of recent work done in the area, ranging
from graph databases to distributed graph processing or graph
partitioning techniques. The latter mainly cope with the problem of
splitting a large graph by assigning its vertices into independent
partitions. While there are several variations of the problem, a
typical objective is to obtain partitions such that the sum of the vertex
weights across partitions is even while the sum of the inter-partition
edges is minimized [
(Messinia,Greece) (Attiki,Attiki)
(Attiki,Messinia) (Attiki,Athens)
(Attiki,Piraeus)
The objective in our work is quite different to vertex
partitioning since we actually do partitioning of the edges of a hierarchical
multigraph. These edges represent interactions between nodes that
need to be investigated according to the imposed hierarchy. In a
typical scenario aggregation of these edges at the higher levels of
the hierarchy tree is more important for the application, while in
certain applications, such as analysis of call detail records,
decision making based on fine-granularity statistics (i.e. low-level
aggregations) is in-fact prohibited by law, so that certain carriers
cannot obtain unfair advantage over their competitors. Our techniques
can benefit systems built for visualizing hierarchical multigraphs
(e.g. [
Assuming a rooted hierarchy tree T , we denote the set of its leaves as Leaves(T ). We refer to a subtree rooted at a vertex x 2 T as Tx. For vertices u and v of T we say that v is descendant of u if there is a path descending from u to v, including the case that u equals v. Vertices u and v are called comparable in T if one of them is descendant of the other and incomparable in T if neither u nor v is a descendant of the other. Also, by depth(u) we shall refer to the number of edges that have to be crossed so as to get from u to the tree root node.
A hierarchical graph is a multigraph G(T ; V; E), where T is a tree, V is the set of vertices in the graph, which are also the leaves of T (V =Leaves(T )) and E is a multiset of edges between the nodes in V . We assume that each instance of an edge is labeled with a unique identifier in order to be able to assign data on them (i.e. in the CDRs example, each edge is associated with a unique key that identifies the corresponding CDR).
In order to represent the set of edges between nodes in T and their relationships, we define the graph GT 2 . Each possible pair of nodes u and v in T is a vertex [u; v] in GT 2 and we refer to u as the source node in T and v as the destination node in T . The edges in GT 2 imply a hierarchy inherited from the hierarchy described by tree T . In particular, there is an edge between nodes [u1; v1] and [u2; v2] in GT 2 if exactly one of the following conditions hold: (i) u2 is a child of u1 in T , (ii) v2 is a child of v1 in T .
Graph GT 2 is used in our framework in order to define our partitioning scheme on the edges of the hierarchical graph. Moreover, the nodes of GT 2 , as will be explained in Section 3.2, are used in order to model possible queries on this data. Then, the edges of GT 2 will determine partitions that contain relevant data for a query.
Each vertex [u; v] 2 GT 2 is a candidate partition Pu;v to be materialized by our partitioning scheme. Keeping all GT 2 in memory is not a feasible solution at its size is quadratic on the size of T . As will be explained, our partitioning process progressively splits the hierarchical graph and constructs a partition tree TP , which is a subgraph of GT 2 . When we need to locate partitions so as to insert new records or answer queries we use TP as our lookup structure. An example partition tree TP which follows the hierarchy of Figure 1 is shown in Figure 2
Finally, by Cu;v we refer to a counter of how many records are contained in the partition Pu;v and by threshold to the value that when the number of records in a partition grows bigger than, the partition has to be split in smaller ones and then get dropped. 3.2
In our framework, retrieval of edges belonging to the hierarchical graph is accomplished via queries that are modeled using the hierarchy tree T . In particular a query Qu0;v0 is denoted as a query edge in T (see Figure 1). This query denotes our intention to retrieve all edges that have as source vertices the leaves of Tu0 and as destination vertices the leaves of Tv0 . When such a query arrives we need to be able to decide which of the materialized partitions in TP may contain relevant graph edges. In order to achieve that we traverse TP in a top-down fashion and check each vertex Pu;v against the query Qu0;v0 . The partition is considered useful in answering the query when their respective source and destination vertices are comparable in T . We continue traversing TP descending the useful partitions until we get to the active partitions (active partitions are these that contain data and are pointed by the leaf nodes of TP as explained in Section 3.3), which are returned to the query for further processing (e.g. filtering of relevant edges). In case that during the traversal we come across a Pu;v for which it stands that u is equal to u0 and v is equal to v0 then we stop the traversal and retrieve the leaf nodes in the subtree rooted at Pu;v. In that case, all the edges in the respective partitions are returned to the user.
In our running example, assuming that the partition tree TP is at the state shown in Figure 2 and that we have to answer query QAthens;Messinia that retrieves all CDRs from locations in Athens to locations in the state of Messinia, we first check the root of TP which is [Greece, Greece]. Since Athens is comparable to Greece and Messinia is comparable to Greece we continue with examining the children of [Greece, Greece], which are [Attiki, Greece] and [Messinia, Greece]. Athens is comparable to Attiki and so is Messinia to Greece, so node [Attiki, Greece] is useful, but Athens is incomparable to Messinia so we do not have to further investigate the node [Messinia, Greece] or any node in TMessinia;Greece. Next we have to check nodes [Attiki, Attiki] and [Attiki, Messinia] which are the children of [Attiki, Greece]. Athens is comparable to Attiki but Messinia is incomparable to Attiki so [Attiki, Attiki] is not considered useful. On the other hand [Attiki, Messinia] is useful since Athens is comparable to Attiki and Messinia is comparable to Messinia. [Attiki, Messinia] is a leaf node in TP and so the traversal ends here and the partition pointed by [Attiki, Messinia] is the result returned to the query. 3.3
A high level description of the partitioning process is as follows: In the beginning let TP consist of just one vertex [r; r] where r refers to the root of T . That means that we initially materialize just one global partition Pr;r containing all possible edges amongst leaves in T . When Cr;r grows greater than threshold the split process is triggered. The split process decides whether Pr;r will be split by its source or destination node, depending on the rules of the chosen partitioning policy, which is discussed later. Each outcome is encoded by a set of nodes that are reachable from node [r; r] in graph GT 2 , depending whether the respective edge denotes a parent-child relationship on the source or destination node.
After the split, the vertices representing the new partitions are added to TP . Vertex [r; r] in TP points no longer to an active partition but we keep it since it describes the records contained in the active partitions pointed by its descendants (the new vertices that we added) and we use this information when we traverse TP in order to insert new records or answer a query. This process takes place for every active partition Pu;v when Cu;v grows greater than threshold after the insertion of new records. This way the vertices in TP that point, or previously have been pointing, to an active partition, form a hierarchical tree. At any moment the leaves of TP point to the active partitions while the inner nodes, including the root, are “aggregations” of these partitions.
Any node in TP can optionally maintain a series of useful
application specific statistics such as the number of records in the
partition, aggregations over measures of these records, calculations
regarding heavy hitters such as top-k sources and top-k
destinations, etc. Furthermore, when fast approximate answers are desired
by the application (for example during exploratory data analysis or
as a preview while the exact answer is computed) it is also possible
to maintain synopses such as Sketches [
In what follows we describe the different split/partitioning policies that we implemented in our system and used in our experiments. First we describe two simple query agnostic policies and next, in more detail, a partitioning policy that we call Query-Driven Partitioning that decides the split to materialize by taking under consideration a set of queries that are most important to the user and makes the split decisions according to them.
Query-Agnostic Policies: The first two policies assume no previous knowledge about the interests of the user. Each of them though utilizes a different heuristic as explained below.
Round-Robin Partitioning: Round-Robin is a simple approach to partitioning the hierarchical graph. Partitions in TP that need to split, are split alternately by source or destination. This process results in creating balanced partitions in the sense that source and destination nodes in a partition have a maximum distance of one hierarchy level. That way the partitions created are not biased towards the source or destination nodes of the constituent edges.
Min-Split Partitioning: Min-Split partitioning policy is a heuristic method, which tries to create the minimum number of new partitions, when an active partition overflows. This policy seems preferable when the goal is to create as few active partitions as possible, while keeping their size close to the selected threshold. Thus, when it has to make a split choice, it simply chooses between the candidate splits the one containing fewer partitions.
Query-Driven Partitioning: In Query-Driven Partitioning we assume that we have a prior knowledge of the queries that the users of the system are mostly interested in. So, when we have to make a split choice we chose the candidate split that suggests a partitioning better suited to answer the set of queries. In what follows we present the idea of Query-Driven Partitioning for hierarchical graphs in a formal way.
Let Pu;v be a partition in the partition tree TP and Qu0;v0 be a query asking for all records having as source and destination all the nodes that are leaves of the hierarchy tree’s T subtrees Tu0 and Tv0 , respectively. Recall that a partition is useful for answering a query if their respective source and destination nodes are comparable in T , otherwise the partition is pruned while navigating the partition tree in search of answers to the query.
For a useful partition, we can define a measure of the overhead that the retrieval of Pu;v adds to the overall cost of answering the query by considering the number of records that belong to Pu;v but are not part of the result of Qu0;v0 . We can use this measure to make the split choice for a partition to be split, by calculating the query answering overhead for each of the candidate splits. Extending this, we can calculate the overhead not just for one query but for a set of queries that the users are mostly interested in.
In order to measure the overhead of a useful partition for a given query, we have to estimate the portion of the partition records that are not useful for the query, but will have to be retrieved when scanning the partition for relevant data. Since both the partition and the query follow the hierarchy implied by the hierarchy tree T we should check the partition’s source and destination nodes u and v against the query’s source and destination nodes u0 and v0, respectively. We will describe the procedure for u and u0, but whatever we mention holds true also for v an v0.
Since nodes u and u0 are comparable (otherwise the partition is not useful), we have to consider the following three cases: (i) u equals u0, (ii) u is a descendant of u0 (depth(u) > depth(u0), and (iii) u0 is a descendant of u (depth(u) < depth(u0)). Let f itness(u0; u) denote the portion of the leaf nodes in Tu that are also leaves in the subtree of Tu0 . In the first and second cases we can safely infer that f itness(u0; u) equals to 1, since Tu is contained in Tu0 . On the contrary, in the third case, Tu0 is contained in Tu and, thus, f itness(u0; u) is calculated by considering the ratio of the leaves of Tu0 over the leaves of Tu. f itness(u0; u) = <>> 1 8 0 , if u and u0 are not comparable , if u and u0 are comparable and depth(u) depth(u0) >>: jleaves(Tu0 )j , otherwise
jleaves(Tu)j Then, the fitness of the partition for the query is computed as: f itness(Qu0;v0 ; Pu;v) = f itness(u0; u) f itness(v0; v) Intuitively, this measure estimates the percentage of records in the partition that are useful for the query, assuming no additional knowledge on the data distribution is given.
In case the partition is split into k sub-partitions, assuming a uniform distribution of the records in Pu;v, then each of these partitions will receive Cu;v of records, where Cu;v is the size of the k partition. Then, given that we have calculated the fitness for each of these smaller partitions denoted as f1; : : : fk, respectively, we compute the overhead of the split as the number of non-useful to the query records expected to be retrieved from the set of partitions belonging to the candidate split as: overhead(Qu0;v0 ; Split(Pu;v)) = Cu;v k
X i=1:::k;fi>0 (1 fi)
For a set of queries the cumulative overhead of the split is computed by summing the estimated overhead for each query. Thus, given a choice of splitting the partition by source of destination, we compare the overheads that we calculated for each of the splits and select to materialize the one with the lowest number. As have
Insertion Job Partitioning Job Query Job client
Metadata Data new data
been explained, our calculations are based on the assumption that the data distribution within the partition is uniform. Of course, this is a bold assumption and we expect that the system may occasionally make wrong decisions, leading to suboptimal splits. An easy workaround is to consider additional statistics, for instance in the form of sketches, that will help better estimate the distribution of records within a partition, at the cost of increased overhead due to bookkeeping of these synopses. We leave exploration of such choices as future work.
We implemented our system as a framework consisting of four main modules, the Buffer Storage, the Admission Manager, the Partitioning Engine and the Query Engine. It uses the well-known distributed data store HBase running on top of HDFS (the Hadoop Distributed File System) and also the Hadoop MapReduce engine in order to accomplish tasks such as loading and repartitioning of the data. We materialize each partition as an HBase table so as to make easier the retrieval of records belonging to one partition by involving the scan of one and only table. Another choice would be to use one big table to store all the records and define each partition space by applying a compound row-key design.
Figure 3 provides a high-level view of the framework. The Buffer Storage module is located at the input of the system and stores temporarily the new records that are streamed from distributed sources, into text files. When the size of these records grows bigger than a maximum buffer size, then the Buffer Storage module sends a message to the Admission Manager module, which executes the task of pulling the records from Buffer Storage and loading them into HBase. After each load of new records, Admission Manager checks for partitions that grew over the partitioning threshold. For any such partition, Admission Manager passes the required instructions to Partitioning Engine, which decides a split according to the selected policy amongst the ones described earlier and applies it.
For the support of these tasks, the system maintains a set of metadata, such as the hierarchy and partition trees that are used, updated and shared by all the different modules. The actual execution of the tasks is taking place as a number of MapReduce jobs, with the most important of them being the Insertion Job which puts each new record in the appropriate partition by looking up the hierarchy and partition trees, and the Partitioning Job which moves the data from a splitting partition to the new partitions.
Finally, we also implemented a Query Engine, which accepts a e itm 25000 g n iiittno 20000 rap 15000 / g idna 10000 o l 5000 0 user query or set of queries and, by looking up the hierarchy and partition trees, discovers the partitions that possibly contain relevant records, scans them and returns the results to the user.
All experiments were conducted on a cluster of seven virtual machines located at the GRNET’s cloud infrastructure Okeanos1. Each machine had 4 processors, 4 GB of memory, while the disk sizes varied from 40 to 100 GB. The operating system installed was Ubuntu Server 12.04 and we used Hadoop version 1.1.2 and HBase version 0.94.6.1. We used one machine as the system master running the NameNode, JobTracker and HMaster daemons, while each of the remaining six slave machines were running instances of the DataNode, TaskTracker and HRegionServer daemons.
In order to be able to generate massive graph data records, we wrote a custom CDR data generator. We used the geopolitical hierarchy of Greece as described by the Ministry of Interior2 to create the hierarchy tree and then we generated a number of phone numbers for each city in it. The resulting hierarchy tree consisted of five levels, which from top to bottom are Country, District, State, City and Phone Number, each of them having a node count of 1, 13, 58, 324 and 1134698 nodes, respectively. The phone numbers consist of regular phone numbers that make and receive calls, inbound-only phones as those found in customer service departments or telemarketing, and outbound-only phones as those used by marketing agencies. Each record in the experiment data sets consists of a source and destination phone number and a unique call record id. Last, we created a random set of 10 queries which we used to guide the Query-Driven policy, but also to examine the performance of each policy in answering them. The queries were picked randomly amongst all possible queries having as source or destination inner nodes of the hierarchy tree, meaning we excluded the root (country) and the leaves (single phone numbers). 4.2
The first experiment we conducted is a comparison of the loading times for each of the partitioning policies we implemented. We created a data set of a total of 50 million CDR records and broke it into an initial ingest set of 20 million records and 10 append sets of 3 million records each. For each of these 11 steps we traced the total time it took to insert the records in HBase and perform 1https://okeanos.grnet.gr/ 2http://www.ypes.gr/ the repartitioning of the schema, when needed. The results of this experiment are summarized in Figure 5. We can see that the MinSplit and Query-Driven policies spent slightly more time in repartitioning the hierarchical graph. This is explained since Min-Split follows a conservative approach of repartitioning the data by taking the minimum number of splits at each step, resulting in more subsequent splits. The Query-Driven policy is often keen to repartition the data by extending the partitioning tree in order to best fit the input queries that drive its selection process. Finally, in Figure 4 we compare the scalability of our framework using a larger number of input data (200 million records) for each policy. Compared to the smaller dataset, the Figure suggest a sublinear increase of the loading times, something that is contributed to the overhead times of the underlying frameworks (e.g. MapReduce jobs setup) that affects more (proportionally) the smaller input. 4.3
In order to examine the quality of the partitions created by each of the policies tested, we propose a measure that we call distance. This metric can be used in applications where the goal is to derive the fewer number of partitions that are each smaller or equal to the selected threshold. Given the actual final partition sizes sizei, where i=1 : : : pm and pm is the number of the active partitions for policy m at the end of the loading phase, we define distancem to be:
vu pm distancem = tuX (sizei i=1 threshold)2 Intuitively, a smaller distance value denotes a set of partitions that are created evenly near, but not exceeding, the selected threshold. We have calculated and present the value of the distance metric for each policy and after each loading/partitioning job in Figure 6(a).
What is worth noticing in this Figure is that the Min-Split policy, while in the beginning was the best amongst the others, later on it created partitions that had a larger collective distance. This is explained by the fact that the choice of the smaller split, leads to many deep splits of the partition tree and, subsequently, when the leaves are reached, the deep partitions that need to be split are getting split in high levels of the hierarchy (on the opposite direction) since this is the only feasible split left. This fact also leads to the increase of the number of active partitions that is evident in Figure 6(b). Thus, even though at each step Min-Split makes (locally) optimal decisions regarding the split that leads to the smaller increase of the distance metric, the final resulting partitioning is worse that the ones achieved by the other policies. 4.4
In order to examine the effectiveness of the dynamic partitioning schema, for each of the policies we ran the set of queries mentioned in Section 4.1. We used these same queries to guide the Query-Driven policy. In Figure 6(c) we can see a comparison of the total records retrieved by each of the policies, and the fraction of them that were useful in answering these queries. As expected the Query-Driven policy has better precision than the other two policies, while the Min-Split policy, which has gone deep in the hierarchy tree, was not able to create partitions suitable for the selected set of random queries.
In this paper we considered the problem of managing big hierarchical graphs by exploiting the implied hierarchy so as to partition the data edges in a way that would better support future retrieval 0 1e+06 0
loading partitioning 1 2 3 4 5 6 7 8 9 10 11
loading/partitioning job (a) Round-Robin round-robin
min-split query-driven 1 2 3 4 5 6 7 8 9 10 11
loading/partitioning job (a) Partitions’ distance 1800 1600 s 1400 ion 1200 itt rpa 1000 fro 800 e m 600 b u n 400 200 0 (b) Number of active partitions and analysis. We evaluated a number of dynamic partitioning policies using open source big data tools on a small cluster of nodes. From the policies considered, the Query-Driven partitioning policy lead to partition schemes that enable faster analysis of the records, assuming that some a-priori knowledge of the user queries is given. In an uncertain environment, the Round-Robin policy seems to result in more balanced partitions with good query performance. Moreover, it has been shown to have the best performance in the loading and partitioning processes.