Francisco Cruz
email: fmcruz@di.uminho.pt

Francisco Maia
email: fmaia@di.uminho.pt

Miguel Matos
email: miguelmatos@di.uminho.pt

Rui Oliveira

João Paulo
email: jtpaulo@di.uminho.pt

José Pereira

Ricardo Vilac
email: rmvilaca@di.uminho.pt

Resource usage prediction in distributed key-value datastores

In order to attain the promises of the Cloud Computing paradigm, systems need to be able to transparently adapt to environment changes. Such behavior benefits from the ability to predict those changes in order to handle them seamlessly. In this paper, we present a mechanism to accurately predict the resource usage of distributed key-value datastores. Our mechanism requires offline training but, in contrast with other approaches, it is sufficient to run it only once per hardware configuration and subsequently use it for online prediction of database performance under any circumstance. The mechanism accurately estimates the database resource usage for any request distribution with an average accuracy of 94%, only by knowing two parameters: i) cache hit ratio; and ii) incoming throughput. Both input values can be observed in real time or synthesized for request allocation decisions. This novel approach is sufficiently simple and generic, while simultaneously being suitable for other practical applications.

Introduction

The ability to predict how a system will behave is critical in Cloud Computing systems. Accurate prediction would allow administrators to make better informed decisions on resource allocation, systems configuration or even the technology to use. Currently, this typically requires extensive testing while still lacking the desirable accuracy levels. This is particularly true for massive scale distributed key-value datastores (often named NoSQL databases), Notably, their highly desirable performance, scalability and availability properties cannot be achieved without careful resource allocation and judicious data placement, which requires extensive testing.

In this paper we demonstrate that, for distributed key-value datastores, it is possible to achieve accurate performance prediction, in real-world scenarios, resorting to only a small fraction of the systems resources. NoSQL datastores make heavy use of buffer caching, specially to improve the performance of read requests. In this work we show that the success of such caching layer is directly related to the datastore's resource consumption and we leverage such relation for resource prediction purposes. In fact, it is known that, for a given throughput, the higher the cache hit ratio, the lower the resource usage of the NoSQL datastore. This is true since each cache hit avoids resource consumption stemming from lower layer access. Moreover, contrary to relational databases, which due to their inherent complexity require more elaborate models, in this work we show that for distributed key-value datastores this correlation is actually enough to accurately predict resource usage of any workload. Such accurate prediction of resource usage then allows system optimization, preparation, and simulation under different conditions. This is particularly important if we aim to effectively deploy NoSQL data stores in the pay-as-you-go model, which is common in the Cloud Computing paradigm. Contributions. a) We provide a mechanism to build a read operation resource usage model and a write operation resource usage model. Both models are hardware dependent, meaning they need to be rebuilt when the hardware changes, but they are generated only once per hardware configuration and can then be used to predict the resource usage for any workload. b) Leveraging these models, we are able to predict a NoSQL datastore resource usage, only by knowing two parameters: i) the cache hit ratio and ii) the incoming throughput. From our experiments using HBase, we accurately predict resource usage for any request distribution and any throughput of read-only and a mix of read and update operations. We achieve an average prediction accuracy of 94%. Roadmap. The rest of this paper is organized as follows. We begin by providing some background about caching mechanisms and NoSQL datatores in Section 2. Section 3 presents evidence on the correlation between the cache hit ratio of NoSQL datastores and resource usage. Section 4 focuses on the prediction of resource usage for read-only operations while Section 5 focuses on write operations. We validate our mechanism using HBase and mixed (read/write) workloads in Section 6, present related work in Section 7 and conclude with Section 8.

Background

Caching mechanisms. Databases make use of buffer caching to improve their read performance. By keeping most frequently accessed data in fast access structures (either implemented by software or hardware) performance can be significantly improved. As a result, the flow of a read request usually takes the following path: i) the client issues a request to read some tuple; ii) the database verifies if the requested tuple is in cache; iiia) if it does the tuple is returned to the client, iiib) otherwise the database tries to fetch the tuple from secondary memory. When using caching one of the main goals is to try to maximize the percentage of requests that are served from cache, also known as the cache hit ratio. A high hit cache rate means that a good number of requests are being served exclusively by the cache, thus avoiding higher CPU and I/O costs from using less efficient storage mediums. When the data size exceeds the cache size, eventually, some data in the cache needs to be removed to give room to more frequently accessed data. This is handled by cache replacement algorithms [START_REF] Puzak | Analysis of Cache Replacement-algorithms[END_REF]. There are several cache replacement algorithms, however one of the most widely used algorithms is the Least Recently Used (LRU) algorithm [START_REF] Sleator | Amortized efficiency of list update and paging rules[END_REF]. Under this replacement algorithm when the cache is full, the algorithm discards data that was least recently used. This algorithm is the one typically used by distributed key-value datastores [START_REF] George | HBase: The Definitive Guide[END_REF] [START_REF]Cassandra -a decentralized structured storage system[END_REF]. Distributed key-value datastores. Distributed key-value datastores run in a distributed setting with tenths to hundreds of nodes, usually composed of commodity hardware. The application data is partitioned and these partitions are assigned to the available nodes according to a data placement strategy. Contrasting with relational database management systems (RDBMS), these datastores only provide a simple key-value interface to manipulate data by means of put, get, delete, and scan operations and they do not offer strong consistency criteria. Complex operations like joining and aggregation are not present and data is denormalized. Considering these characteristics, the success of caching mechanisms is key for performance. In this paper, we focus on HBase which is one of the most successful and widely used key-value datastores [START_REF] George | HBase: The Definitive Guide[END_REF]. Inspired by BigTable [START_REF] Chang | Bigtable: a distributed storage system for structured data[END_REF], HBase's data model implements a variant of the entity-attribute-value (EAV) model and can be thought of as a multi-dimensional sorted map. This map is called HTable and is indexed by the row key, the column name and a timestamp. HBase follows a hierarchical architecture where there is a Master node and there is one or more slave nodes called Regionservers. The row range of a HTable is horizontally partitioned into Regions and distributed over different nodes. Each Region is stored as an appendonly file in the Hadoop Distributed File System (HDFS) [3], whose instances are called DataNodes. Usually, RegionServers are co-located with DataNodes to promote the locality of the data being served by the RegionServer. HBase has a block cache implementing the LRU replacement algorithm. Several key-values are grouped into blocks of configurable size and these blocks are the ones used in the cache mechanism. The block size within the block cache is a parameter but defaults to 64KB.

Interdependence of resource usage and cache hit ratio

Let us consider a server usage metric related to the CPU waiting time on I/O operations (I/O wait ), the time spent on user space (CPU user ) and the time spent on kernel space (CPU system ) in the form: Server usage = I/O wait +CPU user +CPU system . In the following we show the cache hit ratio is effectively related to server usage. To this end, we set up three experiments using a HBase deployment and YCSB [START_REF] Cooper | Benchmarking cloud serving systems with YCSB[END_REF] as the workload generator. These experiments cover a wide spectrum of possible behaviors. With these we are able to show a clear and direct relationship between the cache hit ratio and server usage in NoSQL systems, which lays the foundation for the rest of the paper. Experimental setting: In all experiments, one node acts as master for both HBase and HDFS, and it also holds a Zookeeper [START_REF] Hunt | Zookeeper: Wait-free coordination for internet-scale systems[END_REF] instance running in standalone mode, which is required by HBase. Our HBase cluster was composed of 1 RegionServer, configured with a heap of 4 GB, and 1 DataNode. HBase's LRU block cache was configured to use 55% of the heap size, which HBase translates into roughly 2.15 GB. The Re-gionServer was co-located with the DataNode. The YCSB workload generator ran in a separate node and was configured with a readProportion of 100% (read-only), and with a fixed throughput of 2000 operations per second with 75 client threads. All experiments were set to run for 30 minutes with 150 seconds of ramp up time and the results are the computed average of 5 individual runs. The server usage was logged every second in the RegionServer/DataNode machine using the UNIX top command. The top command gives us the CPU idle metric that is converted to our Server usage metric in the form: Server usage = 100% -CPU idle . By the end of each experiment, we gathered the RegionServer's achieved cache hit ratio. All nodes used for these experiments have an Intel i3 CPU at 3.1GHz, 8GB of main memory, a 7200 RPM SATA disk, and are interconnected by a switched Gigabit network.

First experiment: In this first experiment, a single region was populated using the YCSB generator with 4,000,000 records (4.3 GB). This means that the region cannot be fitted entirely into the block cache: about 1.1 millions records (1.21 GB) remain on secondary memory and must be brought into main memory when requested. There were four different scenarios each with a differently configured request popularity:

1. A uniform popularity distribution, that is all records have equal probability of being requested (the case where the cache hit ratio is minimum); 2. A hotspot popularity distribution, where 50% of the requests access a subset of keys that account for 30% of the key space; 3. A zipf scrambled popularity distribution, highly skewed, but because it is scrambled it means the most popular keys are spread across the key space; 4. A zipf clustered popularity distribution, highly skewed, and clustered, meaning the most popular keys are contiguous, which makes them fall in the same cache block.

The results for this experiment are depicted in Table 1. As expected, the uniform request popularity is the one that achieves the lower cache hit ratio (p hit = 49%) , and thus consumes more server resources (58.35%) while the zipf clustered request popularity has the higher cache hit ratio (93%). This is true because popular keys are found in the same block, which is maintained in memory avoiding cache misses.

Second experiment:

We set up a second experiment, to demonstrate that the behavior observed in the first experiment is independent of request popularities. This experiment is identical to the first one except for the region size, which has now 2,000,000 records (2.14 GB). As a result the region fits entirely into the block cache, thus the expected cache hit ratio is 100%. Third experiment: In this experiment we show that two different distributions, with different data sizes but with the same cache hit ratio, will have the same server resources consumption if subject to the same fixed throughput. We used a similar setting to the first experiment's but changed the number of records of the uniform distribution to 2,141,881(2.3 GB) so its cache hit ratio could also be 93%. The throughput is again fixed at 2000 operations per second. that, for a given throughput, an identical cache hit ratio, regardless of the data size and the distribution results in the same resource consumption.

Correlation between server usage and cache hit ratio: A correlation test using the Fisher's z transformation [START_REF] Fisher | On the probable error of a coefficient of correlation deduced from a small sample[END_REF] with the data from the previous experiments, shows that in fact there is a negative correlation for p-value < 0.001, making it statistically significant. Based on these results, we argue that it is possible to estimate the server usage given the incoming throughput and the cache hit ratio and, in the following sections, we show how this can be done.

Estimating resource usage of read operations

Previous section demonstrated that there is an intrinsic relation between resource usage and cache hit ratio. The cache hit ratio reflects not only the data size, but also the underlying distribution of requests which, in combination with an incoming throughput, corresponds to a given server usage. Furthermore, for a fixed throughput this relation is univocal: for some throughput if two distinct workloads consume the same amount of resources, then they must have the same cache hit ratio. In this section, we show how the server usage of any workload can be estimated simply by knowing its cache hit ratio and incoming throughput. We build on the aforementioned properties to build a tridimensional model, that models the server usage for a NoSQL datastore, when the cache hit ratio and the throughput vary. The objective is to build a model that, for a given hardware configuration, a given hit cache ratio and certain request throughput of an HBase node, allows us to predict the resource consumption of such node. To achieve this we require an initial training step, which is hardware dependent. Consequently, each generated model is only valid for a single hardware configuration but is required to be generated only once. Once we have the model we are able to predict HBase node resource usage for any given workload. As shown previously, request distribution is irrelevant in terms of the relationship between hit cache ratio and resource consumption. Taking advantage of this observation we always consider the uniform distribution in the generation of the prediction models. In fact, such model will still be valid if, on runtime, a different request distribution is observed.

In order to generate the model, our approach is to judiciously choose a number of representative combinations of cache hit ratio and throughput, test them against the desired hardware configuration and then, by using linear interpolation between the different server usage levels measured, we are able to build a tridimensional model that correlates data size, with throughput and expected server usage. Notably, with this approach we are able to achieve very high levels of accuracy. At this point, it is important to note that, for a generic workload generator, it is not possible to define the desired cache hit ratio. Instead we can only set the data size and desired throughput. However, we can take advantage of a simple approach proposed by Che et al. [START_REF] Che | Hierarchical web caching systems: modeling, design and experimental results[END_REF] that provides an estimation without error of the cache hit ratio for the uniform distribution. This way, we can represent the cache hit ratio by its correspondent data size when building the model. Therefore, in the remainder of this section we will mention data sizes implicitly mentioning their correspondent cache hit ratios. Another reason for choosing the uniform distribution is because it allows to reduce the overall training time since it represents the worst case for LRU caches (lowest possible hit ratio for a given data size), thereby the time it takes to populate the data in the NoSQL database is smaller Regarding the process of choosing the representative measures to take, let's begin by looking at an illustrative example. Figure 1 shows the behavior of incoming throughput when data sizes increase for a fixed server usage percentage. Note that, if the data size is smaller than the cache size then only the throughput impacts server usage. In this case, the cache hit ratio is always 100% and the throughput constant for all possible data sizes between 0 and cache size. For larger data sizes, the cache hit ratio drops and cache swapping begins, which in turn means that in order for the server usage to stay the same the throughput must decrease. As a result, this is a boundary point (where data size equals to the cache size). This observation allows us to reduce the number of points to calculate for that section as we just need to build the model from that point onwards. Then, other observations help us choosing the points to measure. For data sizes slightly larger than the boundary point, there is a big drop on throughput in order to resource usage to remain the same. This drop can be more or less abrupt depending on the speed of the secondary memory. In order to capture this behavior in the model we need to increase the number of tested combinations of pairs data size and throughput immediately after the boundary point. Conversely, when the data size is largely increased we can be confident of a long and flat tail, thus not requiring many training points to achieve high accuracy.

The uniform distribution server usage model is automatically generated resorting to a developed Python script and using YCSB as the workload generator 1 . Generally, this script has 2 main parameters: i) a list of cache hit ratios and ii) a list of targeted server usage levels. Hit cache ratios are, as explained earlier, converted to data sizes using the Che's approximation. Then, resorting to a binary search, the script tries to find the necessary throughput of read operations to achieve each specific percentage of server usage for each data size defined as input. Fixing the server usage level and allowing the throughput to be experimentally calculated via the script, allows us to have a representative number of server usage levels without having to test multiple cache hit ratio and throughput combinations in order to have a usable model. When a sufficient number of points for a specific server usage level are found and we resort to interpolation between those points. Namely, using the monotonic spline interpolation of the R project2 embedded into the Python script. This process is repeated for each of the targeted server usage levels. This list does not comprehend all of the possible values between 0 and 100%. Instead, from our experience we noted that a few of them is sufficient (usually 5 equally spaced). Furthermore, by again using linear interpolation between the different server usage levels we achieve very accurate results, and ultimately build a tridimensional model that correlates data size, with throughput and expected server usage.

Model instantiation in our cluster

We ran the automatic server model generator in our cluster using the same setting as the experiments of Section 3. The generated server model is as depicted in Figure 2. There were defined 10 different cache hit ratios: 100%, 95%, 90%, 80%, 70%, 60%, 50%, 40%, 25%, and 15%. These cache hit ratios were then transformed in their data size equivalents to be used as input in the model generator.

The first point is the boundary point corresponding to 2,000,000 of YCSB records. As previously stated, for data sizes slightly larger than the cache size we need to increase the density of points tested to ensure the model is more accurate. Thus, the next point is only a 5% decrease, and the subsequent 6 points are decreases of 10% in the cache hit ratio. On the other hand, predicting a flat long tail from that point on, we just defined 2 points much more apart from each other, 25% and 15% of cache hit ratio, corresponding to 8,000,000 and 12,000,000 records.

In Figure 2 the solid lines correspond to the 5 targeted levels of server usage, namely 80%, 60%, 40%, 20% and 5%. It is general practice in frameworks for automated elasticity of NoSQL datastores [START_REF] Konstantinou | Tiramola: Elastic nosql provisioning through a cloud management platform[END_REF] that the rule governing the addition of new nodes indicate 80% as the maximum usable CPU before a new node is needed in the cluster. This is an empirical higher bound on usable CPU to accommodate operating systems processes, account for possible load spikes and compactions. Therefore, the highest defined level was 80%. When eventually the generator has finished searching for the throughput needed to reach the targeted levels of server usage for the various data sizes, it then interpolates the data that resulted in the represented continuous curves. Finally, we just need to do a final and linear interpolation between these curves. The curves that correspond to the linear interpolation are represented by dotted lines for the server usage levels of 70%, 50%, 30% and 12.5%, which are example levels.

Model accuracy Revisiting the first experiment of Section 3, we can now use the generated model to estimate the server usage for the different distributions. The results are depicted in Table 4. As can be seen, the estimated server usage is almost the same as the observed average server usage, despite all four different distributions with very different cache hit rates. It should be noted that, as expected, the approach predicts the server usage of the uniform distribution with accuracy of 100% due to the similarity between the input usage levels of the model and the ones used in the test. We can also use the generated model to accurately estimate the server usage when the incoming throughput varies. In that regard, we set up two different experiments using the exact same setting as in the experiments of Section 3. For every data point there were 3 independent runs, and the results presented are the computed average. In the first experiment, we populated the HBase instance with 4,000,000 records (4.3 GB). The YCSB's client was configured to use the zipf clustered distribution with 100% read operations, and for a fixed throughput ranging from 250 ops/s to 10,000 ops/s. We also wanted to validate what happens when using a data size not used in the model generator. As a result, we populated the HBase cluster with 3,000,000 records (3.15 GB), and this time using the zipf scrambled distribution, which yields a much lower cache hit ratio (78.8%). As a result, the configured read throughput ranged from 250 ops/s to 7,000 ops/s. The results for each experiment are depicted in Figure 3(a) and in Figure 3(b). They show the estimated server usage compared to the observed one. The estimated results are drawn from our approach using the generated model for read operations, and observing the cache hit ratio as provided by HBase exported metrics. As expected, the estimated server usage in both experiments is very similar to the observed counterpart.

Discussion

The approach described in this section allows to accurately estimate the server usage resorting to an offline trained model based on the uniform distribution.

Using the cache hit ratio and the incoming throughput as the only parameters that affect resource utilization may appear oversimplifying. Specially, when taking account related approaches to usage prediction in RDBMS. However, key-value datastores are fundamentally different from relational databases. In order to attain high scalability, high throughput and high availability, these datastores offer a simple key-value interface based on put and get operations without providing multi record atomic operations nor complex operations like joins and aggregations. On the other hand, RDBMS must cope with a large number of concurrent and lock-prone ACID transactions and need different more complex models for resources, such as CPU, RAM, disk I/O and database locks. These differences allow our simple but effective technique to work. The empirical intuition of why other parameters, such as the I/O costs, do not need to be considered separately is because they are already concealed in the training model. Taking a closer look into the behavior of each distribution in the first experiment, and decomposing the overall throughput into operations hitting and missing the cache, we have:

-Uniform -49% of cache hit ratio; thus 980 ops/s are cache hits, the remaining 1020 ops/s miss the block cache; -Hotspot -56% of cache hit ratio; thus 1120 ops/s are cache hits, the remaining 880 ops/s miss the block cache; -Zipf Scrambled -68% of cache hit ratio; thus 1360 ops/s are cache hits, the remaining 640 ops/s miss the block cache; -Zipf Clustered -93% of cache hit ratio; thus 1860 ops/s are cache hits, the remaining 140 ops/s miss the block cache. By looking at the average resource usage for each distribution, it is obvious that the cost of a cache miss is greater than the cost of accessing the block cache. This implies that the server usage for read operations can be decomposed as the sum of two costs: Usage read = Usage hit + Usage miss The Usage hit is the cost of only accessing the cache, while the Usage miss represents the cost of a miss in the cache. It covers not only the cost of bringing a block into the cache (either from main memory or disk), but also the cost of discarding the least recently used data to make room for the new data block. Thus, when two workloads have identical cache hit ratios and identical incoming throughputs, it means that both workloads have the same number of operations hitting the cache and the same number of operations missing the cache. As a result, once two workloads exhibit the same Usage hit and Usage miss , ultimately exhibit the same server usage.

Estimating the resource usage of update operations

Although workloads are generally dominated by reads, most applications also have updates. We apply a similar approach to update operations. Updates and writes can be used interchangeable, because in key-value datastores, such as HBase and Cassandra, updates and new writes are append-only, so they follow the same write path. Updates in these datastores are first written to main memory before being flushed to disk. Therefore, the resource cost of an update is essentially related with the operation of writing the update to main memory and, from time to time flushing it to secondary memory. As a consequence, contrary to read requests, updates are mostly independent of the request distribution and current data size. In addition, because the write path and the read path in a NoSQL datastore are substantially separated, the overall server usage can be defined as the sum of the usage related with read operations and the usage related with update operations: ServerUsage overall = ServerUsage read + ServerUsage update . As updates are independent of the request distribution and the data size, creating a model to predict the server usage of update operations is simpler than the read model counterpart. The only variable affecting the server utilization is, thus the write throughput. Analogous to the model generator for read operations, we used a Python developed script to generate the server usage model for update operations. It also uses YCSB as the workload generator, but this time configured for updates. As the update model only depends on the throughput, the script has only one main parameter: a list of targeted update throughput points to test. For every element of targeted update throughput there are 3 independent runs, and the server utilization is logged every second in the remote machine where the datastore node is running. When all the defined points are finished, we also resort to interpolation between those points. Like the server model of read operations, the automatic server model generator was used on our own cluster, using the exact same setting. The generated server model for updates is depicted in Figure 4. There were defined 28 different targeted update throughputs from 5 updates per second to 10,000 updates per second. For increased accuracy, the first 10 targeted throughputs fall within the interval of 5 to 1000 updates per second. From that point on, there were 500 increments until 10,000 updates per second, which is the point where the server usage reaches 80%. As can be seen the server utilization for update operations grows linearly with the increased throughput.

Resource estimation for read-write workloads

Along this section we validate our approach using HBase showing that we can accurately predict resource consumption for any given workload even if there is a mix between read and update operations. In the experiments we used the exact same setting as in the experiments of Section 3. For every data point there were 3 independent runs, and the results presented are the computed average. By using both the read and the update model, we are able to estimate the server usage for read operations and update operations independently. However, as described in Section 5 the write path and the read path are mostly separated, thus we are able to estimate the overall server usage just by adding both estimations. In order to validate this assumption, we set up an experiment configured with different read and update mixes, namely: 90% read and 10% update; 80% read and 20% update; 50% read and 50% update; 20% read and 80% update; 10% read and 90% update. The region was populated with 4,000,000 records (4.3 GB) and the requests followed the uniform distribution with a fixed throughput of 118 ops/s, 562 ops/s, 1250 ops/s, 1958 ops/s and 2921 ops/s. These tested throughputs correspond to 5%, 20%, 40%, 60% and 80% server usage levels, as generated by the server model for the uniform distribution. In Figure 5 is depicted the results for this experiment. It shows that our approach is valid and it accurately predicts the server usage even when there are read and update operations simultaneously. However, as seen in Figure 5(c) and Figure 5(d) for the higher values of throughput the observed server usage is higher than the estimated one. These differences can be explained by compactions occurring during the test period that disrupt the readers of records stored on disk. Figure 6 shows the server usage along the entire 30 minute run for the 20% read and 80% update mix (Figure 5(d)) for the 2912 ops/s throughput. Until the compaction process starts (at 1277 seconds) the observed server usage average is the same as the estimated one (44%). Then, the compaction process greatly increases server usage to levels near 100%. When compaction ends regular behavior is resumed. This process greatly impacts the overall server usage average, but even at this point our estimated server usage is only off by 12%, which is the greatest difference observed. It is worth noting, however, that while more powerful hardware and particularly SSDs would attenuate the problem and help improve the estimation, in [START_REF] Ahmad | Compaction management in distributed key-value datastores[END_REF] it is proposed to offload compactions to a dedicated compaction server to prevent the significantly degraded read performance during compactions. Server usage (%) Fig. 6. Observed server usage along a 30 minute run for the 20% read and 80% update mix for 2912 ops/s throughput.

Related Work

A significative group of approaches aims at predicting the resource usage of generic systems such as, virtual machines, thus requiring complex models that must take into account many parameters [27] [26]. As mentioned in the literature, in order to obtain accurate models with fewer variables, it is key to focus on specific applications [START_REF] Jennings | Resource management in clouds: Survey and research challenges[END_REF]. This is the case of performance prediction for RDBMS focused on online transaction processing (OLTP) [22] [21]. Although our work has similarities with the previous approaches, such as resorting to off-line model training, it has different assumptions from RDBMS. These differences significantly change the required approach to accurately predict the performance of key-value datastores. A single resource model is also not achievable for related work that predicts the performance of SQL queries by using models for each database operator (e.g., Sort, Merge Join), which are not present in NoSQL datastores [START_REF] Li | Robust estimation of resource consumption for sql queries using statistical techniques[END_REF]. It is worth mentioning the work on performance prediction for database consolidation, where several database instances are running in the same server and processing different types of workloads and, in many cases, even distinct schemas [1] [8]. Once again, this work needs to deal with the added complexity of RDBMS.

Regarding the techniques used to predict systems' performance, machine learning and analytical modeling are the most commonly used [START_REF] Matsunaga | On the use of machine learning to predict the time and resources consumed by applications[END_REF]. These can be used exclusively or in combination, by resorting to time-series analysis [START_REF] Khan | Workload characterization and prediction in the cloud: A multiple time series approach[END_REF] [13], regression models [START_REF] Desnoyers | Modellus: Automated modeling of complex internet data center applications[END_REF] [28], and clustering [START_REF] Singh | Autonomic mix-aware provisioning for non-stationary data center workloads[END_REF]. These approaches require lengthy training phases to estimate accurately different workload distributions. It is however possible to reduce the duration of this initial phase by using a less accurate model and then refine it, in runtime, with other machine learning algorithms [START_REF] Didona | Enhancing performance prediction robustness by combining analytical modeling and machine learning[END_REF]. Because we target a specific type of system, we are able to reduce our model to only two parameters, the cache hit ratio and incoming throughput. Our approach achieves high accuracy without needing runtime improvements for the model. To the best of our knowledge our approach is the only work that can accurately predict the performance of a NoSQL datastore with a single model. Even if our solution needs offline training, it does not require system traces or runtime mechanisms to improve the precision of the estimation.

Conclusion

Along this paper we focused on a mechanism for distributed key-value datastores resource usage prediction. Our mechanism is able to accurately predict the resource uti-lization for every data size, request distribution and throughput combination. In contrast with previous approaches on prediction systems for cloud environments, we take advantage of focusing on a specific cloud component to improve prediction accuracy and its applicability. In particular, we observed that the majority of the NoSQL systems make use of buffer caching mechanisms to improve performance. Moreover, the effectiveness of such mechanisms is directly related with the performance and, as a consequence, to the resource utilization of the database. This effectiveness can be measured in terms of the hit ratio that the caching mechanism exhibits. The higher the cache hit ratio the more effective the cache mechanism is, and thus more efficient is the database. In this work, we show that a NoSQL workload can be characterized by the incoming throughput and by its cache hit ratio, as the latter is a reflection of the data size and of the distribution of requests. From such observation, we can use the cache hit ratio and the throughput to build a server usage model, that can then be used to predict the resource utilization of any workload only by knowing those two parameters. In our experiments the average prediction accuracy achieved is 94% with a standard deviation of 5.6. Notably, our approach can be effectively used for several practical applications. Examples are automated online load balancing systems, automated resource allocation and even cost-benefit assessment of hardware upgrades to mention a few. In effect, we are currently implementing this mechanism in an automated elasticity tool (MET [START_REF] Cruz | Met: workload aware elasticity for nosql[END_REF]) aiming at improving its load balancing capabilities.

Fig. 1 .

 1 Fig. 1. Typical relation between cache size and throughput for a fixed Server usage .

Fig. 3 .

 3 Fig. 3. Experiments for read-only operations.

Fig. 4 .

 4 Fig. 4. Instantiation of the model for update operations.

  10% read and 90% update.

Fig. 5 .

 5 Fig. 5. Read and update operations mix experiments in HBase.

Table 1 .

 1 Average

	Uniform	49%	58.35%	4,000,000
	Hotspot	56%	46.19%	4,000,000
	Zipf Scrambled 68%	35.91%	4,000,000
	Zipf Clustered 93%	19.28%	4,000,000

Distribution p hit Average Server usage #Records Server usage and cache hit ratio (p hit ) with a region larger than the block cache.

Table 2

 2

	Distribution	p hit Average Server usage #Records
	Uniform	100%	12.29%	2,000,000
	Hotspot	100%	12.14%	2,000,000
	Zipf Scrambled 100%	12.92%	2,000,000
	Zipf Clustered 100%	12.89%	2,000,000

depicts the results. As all data is served only by the block cache, the different request popularities are, as expected, irrelevant to server resource consumption and all distributions use roughly the same resources.

Table 2 .

 2 Average Server usage and cache hit ratio (p hit ) with a region that fits in block cache.

  Table 3 depicts the results that support our claim Distribution p hit Average Server usage #Records

	Zipf Clustered 93%	19.28%	4,000,000
	Uniform	93%	19.76%	2,141,881

Table 3 .

 3 Average Server usage and cache hit ratio (p hit ) results for 2 distributions with different sizes, but with same cache hit ratio.

1

  All the scripts used in this work are openly available at github.com/fmcruz/suhcr/ Instantiation of the server model for read operations based on a uniform distribution.

		35
		30
	)
	Throughput (ops/s x 10 3	10 15 20 25							Server usage 80% 70% 60% 50% 40% 30% 20% 12.5% 5%
		5
		0	0	2.1	3.6	4.2	5.3	8.4	12
				100%	60%	50%	40%	25%	15%
							Data Size (GB)
					Correspondent cache hit ratio (%)
	Fig. 2.

Table 4 .

 4 Observed server usage and Estimated server usage results under four different distributions.

http://www.r-project.org

Acknowledgment

This work is part-funded by: ERDF -European Regional Development Fund through the Operational Programme for Competitiveness and Internationalization -COMPETE 2020 Programme, and by National Funds through the FCT -Fundac ¸ão para a Ciência e a Tecnologia (Portuguese Foundation for Science and Technology) within project POCI-01-0145-FEDER-006961; and project LeanBigData (FP7-619606).


Andrea De Salve

Paolo Mori
email: paolo.mori@iit.cnr.it

Laura Ricci
email: laura.ricci@di.unipi.it

Raed Al-Aaridhi
email: alaaridhi@hhu.de

Kalman Graffi
email: graffi@hhu.de

Privacy-preserving Data Allocation in Decentralized Online Social Networks

Keywords: Decentralized Online Social Network, Data Availability, Privacy Policy, Peer-to-Peer

Distributed Online Social Networks (DOSNs) have been recently proposed as an alternative to centralized solutions to allow a major control of the users over their own data. Since there is no centralized service provider which decides the term of service, the DOSNs infrastructure exploits users' devices to take on the online social network services. In this paper, we propose a data allocation strategy for DOSNs which exploits the privacy policies of the users to increase the availability of the users' contents without diverging from their privacy preferences. A set of replicas of the profile's content of a user U are stored on the devices of other users who are entitled to access the profile according to U's privacy policies. The experimental results obtained from the simulations on traces taken from a real social network show the effectiveness of our approach.

Introduction

A Distributed Online Social Network (DOSN) [START_REF] Datta | Decentralized online social networks[END_REF] is an Online Social Network (OSN) implemented in a distributed and decentralized way. Hence, instead of being based on a single provider which manages the whole system by storing and controlling the data representing users' profiles, a DOSN consists of a (dynamic) set of nodes, such as a network of trusted servers, a P2P system or an opportunistic network, which collaborate to implement the social network services. Therefore, DOSNs shift the control over users' profiles data to the peers that build up the DOSN (i.e., to the users these peers belong to), thus solving some, but introducing new security issues, such as the ones concerning the privacy, integrity and availability of user data.

The privacy of the contents published in users' profiles is one of the main issues in DOSNs because, being personal data, these contents must be properly protected. In particular, DOSN users should be enabled to define their privacy policies, and the DOSN framework is responsible for properly enforcing these policies in order to disclose the users' contents only to authorized friends. On the other hand, DOSNs are also responsible for data availability [START_REF] Mega | On churn and communication delays in social overlays[END_REF], because the contents published in the profile of a user must be kept available even when the owner (i.e., the corresponding peer) is disconnected from the network. Different solutions may be exploited to guarantee data availability in DOSNs, for instance profiles may be saved on a Distributed Hash Table (DHT) [START_REF] Balakrishnan | Looking up data in p2p systems[END_REF] or directly on the peers of the users' friends. Most of the current DOSNs ensure data availability by exploiting a DHT. This implies that contents are stored on untrusted peers and that cryptographic mechanisms are used in order to prevent undesired disclosure of the users data to the owners of the peers storing their profiles. To achieve finegrained access control, every time a user u wants to share a content c with a group of n users (according to the privacy policy defined for c), u encrypts c with a new symmetric key before being stored and, in turn, that symmetric key is encrypted separately with the individual public keys of the n users. Finally, the encrypted symmetric keys are distributed to the n authorized users or directly attached to the content c (see, e.g [START_REF] Baden | Persona: an online social network with user-defined privacy[END_REF][START_REF] Graffi | Lifesocial. kom: A secure and p2p-based solution for online social networks[END_REF][START_REF] Nilizadeh | Cachet: a decentralized architecture for privacy preserving social networking with caching[END_REF]). However, this kind of scheme is not scalable because the number of asymmetric encryption operations and the storage cost of encrypted data depend on the number of users n allowed to read the content c, which is clearly a performance issue because the number of such users can be quite big and can be changed quite often (e.g. addition or removal of friends). As a matter of fact, users tend to have a significantly large number of friends in their networks (e.g. 27% of 18-29 year old Facebook users have more than 500 friend4 ). Recent studies [START_REF] Bodriagov | Encryption for peer-to-peer social networks[END_REF][START_REF] Nilizadeh | Cachet: a decentralized architecture for privacy preserving social networking with caching[END_REF] have investigated the overhead introduced by encryption schemes used in current DOSNs in terms of storage and computational cost by highlighting the impact they have on performance and user experience.

This paper proposes an alternative approach for preserving the privacy of the users' contents and increasing their availability by avoiding the use of data encryption. The approach consists in modelling the contents belonging to the profile of user u using a hierarchical data structure, i.e. a tree, and adopting a proper strategy for allocating the nodes of the tree on the peers of online friends of u. The idea is to allocate a copy of each content c of the tree on another peer that is currently online, and whose user v is allowed to access c according to u's privacy policy. In this way, there is no need of employing encryption mechanisms to protect the confidentiality of u's data once stored on v's peer, because v is entitled to access these data according to u's privacy policy. In fact, v cannot collect additional information about u by directly inspecting the contents of u allocated on its devices (e.g., by browsing the files stored in its file systems), because v would find only the contents that is already allowed to access.

The rest of the paper is structured as follows. Section 2 describes how user's profile is modelled through a tree. In Section 3 we introduce our content replication strategy. Section 4 describes the general system architecture, while Section 5 introduces the main algorithms. Section 6 evaluate the effectiveness of the pro-posed data allocation strategy on a real data set taken from a Facebook application. Section 7 discusses related work, while Section 8 reports the conclusions and discusses future works.

Modelling social profiles

In our system, the social profile P u of a user u is hierarchically modelled by a tree whose nodes correspond to the contents belonging to P u . Since we have a one to one mapping between the nodes of the tree P u and the user's contents, we use interchangeably the terms node or content to refer them. Furthermore, we suppose that an unique identifier is assigned to each node of the user's content tree. The root of the tree is considered as an entry point for all the contents related to a profile owner, such as personal information, interests, friendship information, private communication, posts, images, comments, etc. Each node of the profile tree embeds information about the identifiers of the children and parent nodes and is paired with privacy preferences chosen by the owner. Since we model the profile as a tree, the privileges on each content can be expressed in terms of operations on the tree, such as insertion/removal of a children node and reading/changing the content. Therefore, we can define, for each type of content, four different privileges corresponding to the operations available on the tree: readData, editData, appendChild, and removeChild. The readData privilege corresponds to a read operation, while the other ones correspond to updates of the content or of the tree structure. For instance, a user may exploit the appendChild operation to add comments to a post or to a photo published by the profile's owner. The profile owner specifies the privacy preferences for each node of the profile tree, i.e. the privileges granted to the other users in term of the previous operations. Since we model each user's profile as a tree of contents, a privacy policy that permit readData privilege to a content in the hierarchy implies that the same privilege is also permitted to ancestor nodes. In fact, a property of the hierarchical profile structure is that read permission on one node depends on the privacy policies of the parent nodes. Specifically, the sequence of the privacy policies on a hierarchy of contents can only limit the intended audience In fact, for each node, the user can only specify a privacy policy that restrict the privileges already granted to the ancestor nodes. On the other hand, a privacy policy that permit readData privilege to a node in the hierarchy does not imply that readData privilege is also granted to the descendant nodes in the hierarchy. An example of a user's tree structured profile annotated with the corresponding policies is shown if Fig. 1(a). Since user Charlie can not access the Image1 content, then he can not access the comments, likes and tags of the Image1.

Privacy policies paired with profile's content may exploit different aspects (or attributes) derived from the OSN knowledge. Attributes are used to model interesting properties of users or contents (such as the tagged users, creation date of a content or user's birthday) -as well as the kind of operations performed on resources (such as read or write of contents). shows the replicas table for the profile tree of U where names of the nodes are used as content ids. A primary trusted replica (ptr ) and a set of further replicas (tr ) are defined for each node.

Privacy preserving content replication

DOSNs should guarantee that contents of a user's profile are available at any time in the system. In order to ensure higher data availability, contents of a profile tree are replicated on k peers of the DOSN, called trusted replicas, where k is an input parameter of the system. The idea behind our approach is to exploit the privacy policies paired with the contents of the user's profile to define a privacy-preserving allocation of the profile tree to the available peers of the DOSN. In particular, our approach allocates a content c of the user's u profile tree P u to the peers who are already allowed to read c according to the privacy policy defined by u. In order to select the trusted replicas where the contents of the profile tree P u could be copied, the privacy policy specified for the content c is evaluated by using the authorization component of the privacy-preserving framework and by simulating an access to the content. To this aim, we exploit the privacy-preserving framework we have proposed in [START_REF] De Salve | A privacy-aware framework for decentralized online social networks[END_REF] to support users in the management of their privacy preferences. The framework: i) allows users to define flexible privacy policies to regulate the accesses to the content they have shared by means of a proper Privacy Policy Language, ii) evaluates the privacy policy and returns the corresponding authorization decision that indicates whether or not a user can access the content of another user.

For each content, our framework defines a primary trusted replica (ptr ) and a set of further replicas (tr ). When the user is on-line, it acts as primary replica of its content and it enforces the privacy policies on the content of the profile tree by its own. If the user crashes abruptly or voluntary leaves the social network, the availability of its profile is guaranteed by the other replicas. In this case, the authorization framework is exploited also by the replicas peers to enforce the privacy policy on the users' contents stored on their device. Specifically, when a user v requests access to the content c of P u , the replica storing c evaluates the privacy policies of u (linked to the profile P u ) in order to decide whether to permit or deny the access to c. From a performance point of view, every time a user u wants to share a content c with a group of n users (according to the privacy policy defined for c), our approach avoids a number of encryption operations proportional to n (which, depending on the choosen policy, is typically of the order of number of u's friends or even Friends of Friends). We do not focus in this paper on the problem of ensuring integrity and authenticity of the profiles' contents. To this aim, well-established solutions [START_REF] Aiello | Lotusnet: tunable privacy for distributed online social network services[END_REF][START_REF] Buchegger | Peerson: P2p social networking: early experiences and insights[END_REF][START_REF] Cutillo | Safebook: A privacy-preserving online social network leveraging on real-life trust[END_REF] could be adopted.

The system model

In this section we describe the general architecture of the system we propose to support the privacy-aware allocation of the user's contents. In the following, we assume a one-to-one mapping between users and their peers and we use interchangeably the terms peer or user to indicate them.

Each user u is bind to its user descriptor D u which contains information about the IP address of the corresponding peer, the online/offline status of the peer, the identifier of the root of its profile tree and the current replicas available for all contents of the profile. Since the descriptor D u must be available to all the peers which are going to access the profile of u when it is offline, it is stored on a DHT and may be retrieved by exploiting the identifier of the u. Note that the DHT is exploited not only to keep track of the content replicas, but it is also indispensable for peer bootstrapping, addressing and for supporting the search of new friends.

The replication framework

When a content is created, the user who created it specifies the privacy policy for that node of the profile tree that best matches its needs. Such policies are statements of the privacy policy language that specify who has access to the content in terms of a set of features encoded by attributes.

The association between the contents of a profile tree P u and the trusted replica peers where they are stored are maintained in the replicas table R u (see Fig. 1(b)) which is located in the user's descriptor D u of the profile owner. The replicas table R u contains, for each content c of P u , the trusted replica list R u (c) = {ptr, tr 1 . . . , tr k-1 }, where ptr is the primary trusted replica, while tr 1 . . . , tr k-1 } are the other replicas of the content. For maintaining replication transparency, the trusted replica lists are managed according to a passive replication model [START_REF] Ghosh | Distributed systems: an algorithmic approach[END_REF] where every user communicates only with ptr, the primary trusted replica.

When the peer of the profile owner is online, it becomes primary trusted replica of their contents. Primary trusted replica peers are responsible for the availability of the contents, for enforcing privacy policies every time a user tries to access a content and for the selection of the new trusted replicas. Only the primary replica can elect new replicas for the content and it can add them at the end of the trusted replica list. The selection of a new trusted replica for a content c of user profile P u can be performed: i) actively by the content owner u during the online periods since it acts as primary replicas for their contents or ii) by the current primary trusted replica of c, when the owner is not online.

Let us denote by tr(c) and ptr(c) respectively, the set of trusted replicas and the primary trusted replica for the content c and assume that user t wants to read a content c of the profile P u of a user u which is offline. t navigates through the profile of u starting from the root of the profile tree, whose reference is stored in the descriptor D u (we recall that D u contains a reference to the replica owning the root of the profile). Recall that each node (content) of the profile stores the identifier of its children in the tree. The access to a content c thus requires a sequence of recursive accesses to the descriptor stored on the DHT to obtain the references to the replicas storing the nodes on the path from the root of the DHT to the requested content. When the content c has been located, t accesses the replicas table R u for that content stored in D u , and locates the trusted replicas of c (see Fig. 1(b)). If a primary trusted replica is available, t sends an access request to it. The primary replica checks whether t is authorized (according to u's privacy policy) to read the content c and if the case, it send back to t the requested content. Robustness against peers failure and involuntary disconnections of replica peer may be ensured through periodical exchanges of heartbeat messages between the primary trusted replica and the trusted replicas.

The Algorithms

In this section we describe the algorithms executed by the peers in order to realize the contents allocation strategies described above. Initially, we assume that each user is online and connected to the system. When a peer u leaves the DOSN, it executes Algorithm 1. Initially, u retrieves its user descriptor D u from the DHT and notifies its status to the other peers by updating its availability information (line [START_REF] Baden | Persona: an online social network with user-defined privacy[END_REF][START_REF] Balakrishnan | Looking up data in p2p systems[END_REF]). At the moment of disconnection, u has to update the trusted replica list tr of each content c it stores, by removing its contact information (line [START_REF] Buchegger | Peerson: P2p social networking: early experiences and insights[END_REF][START_REF] Cutillo | Safebook: A privacy-preserving online social network leveraging on real-life trust[END_REF][START_REF] Datta | Decentralized online social networks[END_REF][START_REF] De Salve | The impact of users availability on on-line ego networks: a facebook analysis[END_REF][START_REF] De Salve | A privacy-aware framework for decentralized online social networks[END_REF]), regardless of the user n to which the content c belongs. In this way, u will no longer be considered trusted replica for the contents while the replica at the first position of the in the list tr (if any) will act as the primary trusted replica for the content. For availability purposes, the trusted replica u which is going to disconnect from the network, must keep a copy of each stored content c until its reconnection. Finally, u can leave the underlying DHT network (line 11).

Algorithm 2 specifies the steps performed by a user's peer u who wants to join the network. Initially, u retrieves its user descriptor D u and updates its status information in order to inform other peers about its presence (line [START_REF] Baden | Persona: an online social network with user-defined privacy[END_REF][START_REF] Balakrishnan | Looking up data in p2p systems[END_REF]). By using the user descriptor D u , u will be able to get its replica table R u and to have information about their contents and the replicas available for each of them. It may happen that the peer u that joins the system is an old trusted Algorithm 1 User u leaves the network leaveDHT(); 12: end procedure replica peer that reconnects to the DOSN and has old contents stored on its local memory. For each content c of a generic user n stored in the local memory of user u two different scenarios may occur. When there are trusted replicas available for the content c (line 9), then the peer u has to synchronize the local copy of the content with those currently available in the system by sending c to the current primary replica ptr(c) (line 10). In the case where u is the owner of c, u becomes the primary trusted replica of c otherwise it can remove the local copy content (line [START_REF] Graffi | Lifesocial. kom: A secure and p2p-based solution for online social networks[END_REF][START_REF] Kumaraguru | A survey of privacy policy languages[END_REF][START_REF] Mega | On churn and communication delays in social overlays[END_REF]). When there are no trusted replicas for the content c, the user u will act as primary trusted replica for c and provides a (possibly) outdated copy of c in the system (line [START_REF] Nilizadeh | Cachet: a decentralized architecture for privacy preserving social networking with caching[END_REF][START_REF] Sharma | Supernova: Super-peers based architecture for decentralized online social networks[END_REF][16]). When a user's peer decides to behave as a trusted replica for a content c of the profile P n , it needs to modify the replica table R n by adding the contact information of the new trusted replicas (line 12 and 15). It is worh noting that after the execution of the Algorithm 2 all the contents stored on the local memory of u have been either updated or provided by u.

At the end of the join procedure, a periodic dataAvailability operation is initiated.

When a peer u becomes a primary trusted replica for a content c, it has to periodically check if new trusted replicas are needed and, in this case, it has to find another peer who is currently online and is allowed to access c according to u's privacy policy. Algorithm 3 describes the periodic steps performed by primary trusted replicas in order to ensure that at least k trusted replicas are available for each content they store. Every interval A time unit, a primary trusted replica u selects a content c of the profile P n stored on its local memory and checks whether the number of available trusted replicas for c is less then k (line 3-7). In this case, u executes an election procedure which selects a new trusted replica for c. User u gets the set of online users F having a friendship relation with n (the owner of the content) and then use the authorization module of the privacy preserving framework to evaluate whether a friend f ∈ F is authorized to read the content c of n (line 8-15). The privacy policy of c is evaluated for each user f in the set of neighbours by simulating an access request on a user's content Algorithm 2 User u join the network start dataAvailability(u,Du); 19: end procedure c and it may only return permit or deny (line 11). The set of feasible trusted replica peers where the unencrypted user content c could be copied is composed of the users who have obtained a permit authorization decision (line 12-14). Finally, the data availability procedure chooses a trusted replica peer, among those available, that meets specific performance objectives (line 16). If such user exists, the content c is copied in clear on its local memory, which is considered a trusted replica (line 17).

Experimental results

With the aim of evaluating the feasibility of our approach, we have developed a set of simulations of our system using the P2P PeerfactSim.KOM5 simulator, a highly scalable simulator written in java. We have also implemented a Facebook application, called SocialCircles!6 which exploits the Facebook API to retrieve the following sets of information: i) the friendships and profile information of registered users and ii) we sampled the Facebook chat status of registered users and their friends every 8 minutes for 10 consecutive days (from Tuesday 3 June 2014 to Friday 13 June) in order to derive the average session length of the users [START_REF] De Salve | The impact of users availability on on-line ego networks: a facebook analysis[END_REF].

We exploit some reference policies to evaluate our framework. Although the enforceable privacy policies are expressed by using proper Privacy Policy Language [START_REF] Kumaraguru | A survey of privacy policy languages[END_REF], for the sake of clarity, we express the policy examples in natural end while 20: end procedure language. We use policy's attributes to model friendships, common friends number, geographic location, common interests and the strength of the relationship in terms of Dunbar circles, which is a representation of the intensity of the relationship between two users which can be approximated by using the number of interactions occurred between users [START_REF] De Salve | The impact of users availability on on-line ego networks: a facebook analysis[END_REF]. Consider the user Alice and a content c of her profile. In the experiments, we consider the following reference policies: The experiment size is set to 3000 peers and contains a subset of the Social-Circle! Facebook dataset. From minutes 1 to t 0 = 200, the simulation is initiated, each peer joins the DHT and then loads its attribute values. Those users which represent a user registered to our Facebook application start to create and publish an empty profile which can be used by their friends to access Posts and Images of the profile. In our simulation 35 users in total publish their profiles, in this phase. Afterwards, user churn is activated and around 50% of the peers leave the network (see Figure 2(a)). We used a exponential churn models (which simulates the temporal absence of hosts) based on the realistic measurements obtained from our dataset [START_REF] Mega | On churn and communication delays in social overlays[END_REF] while the network layer is implemented by the Global Network Positioning (GNP) module based on measurements from the PingER project.

At the time t 0 the set-up phase is finished and those peers which represent a user registered to the Facebook application start to publish either Posts or Images with a probability of 0.5. Furthermore, the data availability protocol (Algorithm 3) is started. In Figure 2(b) we present the number of user's contents created during the simulation. A total of 600 data objects were created during the simulation. The number of profiles in the DOSN is constant and equal to the number of registered users, while Posts and Images are generated with equal probability and do not exceed the number of 300 contents. A the moment of the creation of a content, user assign to the generated content a privacy policy randomly chosen among those previously defined, namely: i) Policy 1, ii) Policy 2 with a number of common friends (k) equal to 8, iii) Policy 3 with Dunbar circle C randomly chosen, v) Policy 4 and vi) Policy 5 with distance d equal to 5 km. The Figure 2(d) shows the number of contents assigned to each privacy policy.

During the simulation, users publish their contents and select at most k trusted replicas to increase the availability of each content. We investigated the availability provided by our approach by selecting the maximum number of replicas k for each content equals to 2,4 and 8. In Figure 2(c) we present the total number of online trusted replicas available during the simulation. As we use a passive replication model where the copies of each content are managed by a primary trusted replicas, we differentiate primary trusted replicas (Primary) from other trusted replicas (Tr). The number of primary replicas created during the simulation is almost 100 and the same for all values of k, while the total number of online trusted replicas depends on the value of k. Furthermore, we focus on the costs of the data availability service in terms of message consumption and network traffic and for diffent values of maximum number of replicas k. Figure 2(e) and 2(f) present the amount of system messages and the number of bytes generated by the proposed algorithm. The total number of messages exchanged in a given time-interval is proportional to the number of online users in the systems. In order to investigate the rate of available contents in the network we start reads to the published profiles contents every 5 minute. Access request to the contents may return permit (i.e. user can read the content) or deny (user can not read the content). Figure 3(c) shows the number of reads operation performed by online peers. On the basis of the privacy policy defined for the requested content, the applicant could have or not the right to access it. In order to assess the availability of our approach, we compute the number of contents available in the system through trusted replicas by counting the number of times a content is replicated at least on one online trusted replica. The Figure 3(a) shows the number of available/unavailable contents for different values of maximun number of replica k. It is important to note that our approach guarantees the availability of the most part of the contents present in the system by selecting as replicas the peers which can store unencrypted copy of the contents. During the simulation, an average number of 100 contents (out of 600) are not available in the system. Since the contents of a profile tree can be accessible only by authorized user (if any), we have measured the fraction of unavailable contents that can be accessed by at least one online user in the system. This metric (named real unavailale) is almost the same for all the values of k and it proves that the real amount of unavailable contents that can be accessed by online users is at most 10. As shown by the graph in Figure 3(a), the maximun number of trusted replica k selected for each content does not strongly affect the availability of the system because, over time, contents are incrementally replicated on the set of peers of the users who can access them. However, higher values of the maximum number of trusted replicates k increase the robustness of the system against failures. We investigated the load of replicas by measuring the average number of contents stored on each trusted replicas (avg) along with the standard deviation (stdDev). The results given by Figure 3(b) indicate that our proposed data availability protocol balances the load uniformly between the different trusted replicas. The average load is about 2 content stored on each trusted replica for k equals to 4 and 8 while the standard deviation clearly shows that the most part of replica peers store in their local memory less then 6 contents. As shown by the Figure 3(c), the selected trusted replicas properly enforce the privacy policies by allowing access only to authorized users. Finally, we assessed the feasibility of our approach by measuring the time consumed by each operation. Figure 3(d) shows the average duration (with 95% C.I.) obtained during the simulation. The average duration of operations required by our protocol is quite low and equal to 10 seconds, regardless of the maximun number of replicas k.

To the best of our knowledge, none of these previous works take into account users' privacy policies to drive the allocation of the data. In order to help users in protecting their personal content, current DOSNs adopt simple privacy policies based on a combination of encryption techniques, namely Public Key Infrastructure (PKI) and Attribute Based Encryption (ABE). In Diaspora7 users may agree to act as a local server in order to keep complete control of their data, or choose to use an existing server. The Diaspora server grants its administrator read and write access to unencrypted users' information hosted by its device. Users rely on the server's owner to maintain the security, integrity, and reliability of their data. In Safebook [START_REF] Cutillo | Safebook: A privacy-preserving online social network leveraging on real-life trust[END_REF] users associate a particular trust level with their friends and this level is used to select closely related contacts that primarily will store the user's data. The trust level of the nodes is directly specified by the users without taking into account their privacy policies. The user's data are allocated by using an access control scheme based symmetric and asymmetric encryption. PeerSoN [START_REF] Buchegger | Peerson: P2p social networking: early experiences and insights[END_REF] exploits local users devices to store their data securely. Users data are encrypted with the public keys of the users who have access to it. Lotus-Net [START_REF] Aiello | Lotusnet: tunable privacy for distributed online social network services[END_REF] relies on structured architectures where users' data could be stored and replicated securely by combining symmetric and asymmetric encryption. In Su-perNova [START_REF] Sharma | Supernova: Super-peers based architecture for decentralized online social networks[END_REF] users' data are kept available by exploiting stranger or users' friends peers. The user's data on stranger peers (or on friend peers) are encrypted by using a threshold-based secret sharing approach. In LifeSocial.KOM [START_REF] Graffi | Lifesocial. kom: A secure and p2p-based solution for online social networks[END_REF] users' data could be stored and replicated on different peers arranged according to a DHT. Any data is first encrypted with a new symmetric key, then, this symmetric key is encrypted individually with the public key of the users able to access the data and attached to the user's content.

In Cachet [START_REF] Nilizadeh | Cachet: a decentralized architecture for privacy preserving social networking with caching[END_REF] users' data are securely stored and replicated on the peers of a DHT by using a cryptographic hybrid structure which leverages Attribute Based Encryption (ABE) and symmetric key. Only users that meet the policy can decrypt the private key used to encrypt the content. Persona [START_REF] Baden | Persona: an online social network with user-defined privacy[END_REF] leverages a hybrid architecture where users store their data either on their local storage device or on external storage services provided by others parties. Private user data in Persona is always encrypted with a symmetric key.

In turn, the symmetric key is encrypted with an ABE key or with a traditional public key.

Conclusion and Future Works

This paper proposes a content allocation strategy supporting contents availability in a Distributed Online Social Network. The main feature of this allocation strategy is that it tries to replicate the content of user u without encrypting it on the peers which are considered trusted, i.e., the peers belonging to users who can access the content of u according to u's privacy policy. The experimental results performed on real data traces show that the proposed approach ensures high availability of the profiles' contents.

We plan to extend our system in several directions. The partitioning of the tree can be optimized in order to store contents belonging to sub-paths of the tree on the same replica so that the number of DHT accesses during the visit of a profile is minimized. Finally, even if our approach guarantees an high degree of availability, we are working on a mechanism to store a content on the DHT when there are no available trusted replicas.

Fig. 1 .

 1 Fig.1. Fig.1(a)shows the general profile tree data structure which contains posts and images of the user U. The root of the profile tree and the Images node are intended to be shared with the entire circle of U 's friends. Finally, the privacy policies specified by U for the contents Image1 and Image2 allow access to the set of users {Alice, Bob} and {Alice, Bob, Charlie}, respectively. While Fig.1(b)shows the replicas table for the profile tree of U where names of the nodes are used as content ids. A primary trusted replica (ptr ) and a set of further replicas (tr ) are defined for each node.

Policy 1

 1 Only users who have a friendship relationship with Alice can read c. Policy 2 Only users who have a friendship relationship with Alice and at least k common friends with Alice and can read c. Policy 3 Users who have a friendship relationship with Alice can read c provided that they are in a specific Dunbar circle C. Policy 4 Only users who have a friendship relationship with Alice and common school information can read c Policy 5 Only users who have a friendship relationship with Alice and a home location which is far at most d km from Alice's home can read c.

Fig. 2 .

 2 Fig. 2. Contents and trusted replicas created during the simulation.

Fig. 3 .

 3 Fig. 3. Assessment of contents availability and load.

  Algorithm 3 Periodic actions performed by primary trusted replicas u

	Require: k -max number of trusted replicas
	1: procedure dataAvailability(u,k)
	2:	while intervalA do
	3:	get c ∈ Local memory | ptr(c) = u;
	4:	n=owner(c);
	5:	Dn =getUserDescriptor(n);
	6:	tr=getTrustedReplicas(c);
	7:	if tr < k then	looks for new trusted replica
	8:	F =getOnlinePeer(Dn);
	9:	candidates = ∅;
	10:	for f ∈ F do
	11:	result=evaluateAccess(c,READ,f );
	12:	if result = PERMIT then
	13:	candidates = candidates ∪ {f };
	14:	end if
	15:	end for
	16:	r=selectTrustedReplica(candidates);
	17:	tr(c) = tr(c) • r;	trusted replica selection
	18:	end if
	19:

Statistics of Facebook are available at http://pewrsr.ch/1dm5NmJ

Available at www.peerfactsim.com/

Available at http://www.socialcircles.eu/

https://joindiaspora.com/


Fábio Coelho
email: facoelho@inesctec.ptjop

José Pereira

Ricardo Vilac ¸a

Rui Oliveira

Holistic Shuffler for the Parallel Processing of SQL Window Functions

Window functions are a sub-class of analytical operators that allow data to be handled in a derived view of a given relation, while taking into account their neighboring tuples. Currently, systems bypass parallelization opportunities which become especially relevant when considering Big Data as data is naturally partitioned. We present a shuffling technique to improve the parallel execution of window functions when data is naturally partitioned when the query holds a partitioning clause that does not match the natural partitioning of the relation. We evaluated this technique with a non-cumulative ranking function and we were able to reduce data transfer among parallel workers in 85% when compared to a naive approach.

Introduction

Window functions (WF) are a sub-group of analytical functions that allow to easily formulate analytical queries over a derived view of a given relation R. They allow operations like ranking, cumulative averages or time series to be computed over a given data partition. Listing 1.1 presents one window function that is expressed in SQL by the operator OVER, together with a partition by (PC) and an order by (OC).

Listing 1.1: Window Function example.

s e l e c t r a n k ( ) OVER( P a r t i t i o n By A Order By B) from R The growing Big Data trend is shifting the processing of these functions to cloud environments deploying computation to a distributed mesh of computing nodes, where data and processing are naturally partitioned. The distributed execution of queries leverages on data partitioning as a way to attain gains associated with parallel execution. Nevertheless, the partitioning strategies are typically governed by a primary table key, which only benefits the cases where the partitioning of a query matches that same key as depicted in Figure 1. When the query has to partition data according to a different attribute in a relation, it becomes likely that the members of each partition will not all reside in the same node, which compromises the final result for a subgroup of non-cumulative analytical operators, such as rank, since all members of a distinct partition need to be handled by a single entity, in order not to incur in unnecessary sorting steps, which is the most costly operation [START_REF] Cao | Optimization of analytic window functions[END_REF].

In this paper we propose an Holistic shuffler which according to the partitioning considered by the ongoing window function will instruct workers to handle specific

1 1 1 1 1 1 2 3 1 2 1 1 A B PK 2 2 2 2 1 2 2 3 1 3 3 1 A B PK 3 3 3 3 2 2 2 2 1 1 3 3 A B node #2 node #3 node #1
select rank() OVER (partition by PK) from table Fig. 1: Data partitioning according to the primary key (PK). partitions according to the data sizes they hold, minimizing data transfer among workers. We present the design and action of the shuffler which is based on prior knowledge of the data size distribution of each column in the relation, reflecting the data size held in each partition rather than considering the actual data value as seen in common use of database indexes. The preliminary evaluation of our mechanism shows that our approach is able reduce data transfer by 85% when compared with a naive approach.

Roadmap: In the remainder of this paper, Section 2 introduces the distribution considered and Section 3 presents the design and architecture of the Holistic Shuffler we propose. Section 4 accesses our approach. Section 5 briefly reviews related work and overviews our contributions.

Data Transfer Statistics

The use of indexes [START_REF] Garcia-Molina | Database systems: the complete book[END_REF], histograms [START_REF] Poosala | Approximate query answering using histograms[END_REF] and other heuristics are now a staple feature in modern database systems, as they allow to expedite operations, avoiding full scan operations over relations. More recently, these strategies started to be present in cloud infrastructures [START_REF] Chen | A framework for supporting dbms-like indexes in the cloud[END_REF], allowing for processing on primary and secondary attributes. Histograms are commonly used by query optimizers as they provide a fairly accurate estimate on the data distribution, which is crucial for a query planner. An histogram allows to map keys to their observed frequencies. Database systems use these structures to measure the cardinality of keys or key ranges. Without histograms, the query planner would have to assume uniform distribution of data, leading to incorrect partitioning, particularly with skewed data [START_REF] Poosala | Improved histograms for selectivity estimation of range predicates[END_REF], a common characteristic of non synthetic workloads.

When a query engine has to generate parallel query execution plans to be dispatched to distinct workers, each one holding a partition of data; having histograms like the aforementioned ones is an asset, but it does not completely present an heuristic that could be used to enhance how parallel workers would share preliminary and final results. This is so as such histograms only introduce and insight about the cardinality of each partition key. In order to improve bandwidth usage, thus reducing the amount of traded information, the histogram also needs to reflect the volume of data existing in each node, instead of just considering the row cardinality.

Histogram Construction

The cornerstone of the contribution we present is based on merging the knowledge for row cardinality and average row size for each partition. Both could be seen as global metrics that a given query engine may be able to produce and maintain, as this type of information is already used for similar purposes. The cadence at which the histogram should be updated was left outside of the scope of this paper due to space constraints. Nevertheless we note it as a relevant topic, since the optimal performance of any heuristic based approach is entirely connected with its own representativity. Figure 1 presents the result of hash partitioning a relation in 3 workers, according to key PK. The histogram to be built will consider the cardinality of each value in each attribute of the relation for each single partition. Since the construction of the histogram should not be done during query planning time, it cannot know beforehand the partitioning clauses induced by queries. As such, we consider all distinct groups of values in each attribute. Each partition will contribute to the histogram with the same number of attributes as the original relation, plus a key, reflecting the data in that partition. Afterwards, each worker should be able to share its partial histogram with the remainder workers in order to produce the global histogram. foreach key : attr 3:

count ← count(distinct) 4:

size ← size(key) 5:

hist P n (key, attr) ← (count, size) 6: P n ← [attr 1 , attr 2 , attr n ] 7: hist P n ← [key, attr 1 , attr 2 , attr n ] 8: function GLOBAL HISTOGRAM(P n ) 9:

for each attr : P n 10:

COUNT DISTINCT KEYS(attr) Algorithm 1 governs how each partition histogram (hist P n ) should be built. Briefly, each attribute (attr) is traversed and for each key, the total number of distinct occurrences of that key is computed, together with its size. The pair of values is then added to the histogram. The tables in Figures 2(a), 2(b) and 2(c) present the resulting histograms for each partition according to Figure 1.

When all workers have completed computing the histogram regarding its own physical partition, they need to share it with a designated worker, so that the global histogram is also computed. The global histogram will traverse each physical partition histogram and evaluate, for each key, which is the physical partition that holds the largest volume (in size, evaluating the cardinality × average row size). The table in Figure 2(d) depicts the final result of the global histogram. The global histogram will have the same number of attributes of each partition histogram. Please note that the keys for both the physical partition and the global histograms are not the primary keys of the relation, but rather the distinct values found in each attribute during the construction of each partition histogram. Therefore, we provide a brief example on how to read this histogram. Consider that a given query requires data to be partitioned according to attribute A. Then, the histogram informs that key 1 and 2 have the largest volume of data respectively in partitions p1,p3 and, and regarding key 3, partitions p2 and p3 both hold the same volume.

Holistic Shuffler

The Holistic Shuffler leverages on the data distribution data collected by the Global Histogram in order to expedite shuffling operations. The Shuffle operator can be translated into a SEND primitive that forwards a bounded piece of data to a given destination, considering the underlying network to be reliable. During the workflow for processing a window operator, there are two different moments where data needs to be shuffled. The first moment occurs immediately after the operator starts, and its goal is to reunite partitions, thus fulfilling the locality requirement. The second moment occurs in the end of the operator and it is intended to reconcile partial results in order to produce the final result.

Both operators define distinct goals regarding the destinations that need to be chosen for each forwarding operation. Therefore, we establish two shuffle operators, the local shuffle and the global shuffle contemplating each set of requirements. The Local Shuffle will dispatch rows of a given partition to the worker responsible for that partition as dictated by the Global histogram. Algorithm 2 depicts the behavior of the operator. As each row is read from scanning the partition, the value contained in that row for the attribute that dictates the partition clause is collected (partition). This value is then used together with the partitioning attribute to obtain the destination worker from the global histogram. If this row is not meant to be handled by the ongoing worker, then it is forwarded to the correct worker. if worker id = master worker then 7:

Algorithm 2 Local Shuffle Operation

SEND(master worker, row)

The Global Shuffle will forward all aggregated rows to the worker that will hold the overall largest data volume, which will from now on designate as master worker. By instructing the workers that hold the least volume of data, we are promoting the minimal usage of bandwidth possible. Algorithm 3 reflects the behavior for the Global Shuffle operator. The input data considered by the Global Shuffler is composed by the ordered and aggregated rows, both produced by earlier stages of the worker work flow. Such rows will now have to be reconciled by a common node, which for this case will be dictated by the master node, as previously stated. Upon start, the Global Shuffle operator will interrogate the histogram regarding the identity of the master node. Afterwards, as each aggregated row is handled by the operator, it is forwarded to the master worker, if the master node is not the current one.

Evaluation

Along the current Section, we present the preliminary evaluation for the contributions we propose. In order to evaluate our contribution, we used RX-Java [START_REF]Reactive programming for java[END_REF] to simulate the parallel execution of the window operator in several workers. This framework establishes bindings to the Java language, enabling it to use the semantics of Reactive Programming [START_REF]Reactive programming[END_REF]. We selected this framework as it allows to establish a series of data streams, mimicking the window operator data flow. Throughout the evaluation, we employed a single ranking query (listing 4) holding a window function over a syntheticallygenerated relation as in TPC-C's Order Line relation, holding 10 distinct attributes.The values considered for each of these attributes were distributed according to TPC-C's specification. The generated data composes 100 distinct partitions, each one with 500 rows. Globally, the Order Line relation held 500 K tuples. The experiments were performed on a system with an Intel i3-2100-3.1GHz 64 bit processor with 2 physical cores (4 virtual), 8GB of RAM memory and SATA II (3.0Gbit/s) hard drives, running Ubuntu 12.04 LTS as the operating system. For comparison purposes, we report the results by using a naive approach and our Holistic Shuffler. The naive approach, instead of using any knowledge to forward data, disseminates all data among all participating workers. The results in both pictures are depicted according to a logarithmic scale, in the average of 5 independent tests for each configuration.

The Holistic technique we propose required in average only 14.7% of the rows required for the Naive approach to reunite all the partition in each computing node, as depicted in Figure 3(a). The large difference is justified by the fact that the naive approach reunites partitions by forwarding data among all participating nodes, which intrinsically creates duplicates in each node, growing in proportion to the number of nodes. The Local Shuffling stage is depicted in Figure 3(b), in which we varied the number of computing nodes that participate in the computation of the ranking query, verifying the number of rows that were forwarded according to each technique.

Related Work and Conclusion

Despite its relevance, optimizations considering this operator are scarce in the literature. The work by [START_REF] Cao | Optimization of analytic window functions[END_REF] or [START_REF] Zuzarte | Winmagic: Subquery elimination using window aggregation[END_REF] are some of the exceptions. Respectively, the first overcomes optimization challenges related with having multiple window functions in the same query, while the second presents a more broad use of window functions, showing that it is possible to use them as a way to avoid sub-queries and reducing execution time down from quadratic time.

In this paper we proposed an Holistic Shuffler, tailored to be used for the efficient parallel processing of queries with non-cumulative window functions. The design is based on a statistical method that can be used to reduce the amount of data transfered among computing nodes of a distributed query engine, where data is naturally partitioned. Moreover, the preliminary evaluation we present shows that by applying this methodology we were to reduce in average 85% of data transferred among computing nodes. As future work, we plan to translate this approach to a real query engine.

Fig. 2 :

 2 Fig. 2: Partition and Global Histogram construction.

Algorithm 1

 1 Histogram Construction in Partition n 1: procedure COUNT DISTINCT KEYS(attr) 2:

  s e l e c t r a n k ( ) OVER ( p a r t i t i o n by OL D ID o r d e r by OL NUMBER) from Order L i n e

Fig. 3 :

 3 Fig. 3: Comparison results between the Naive and Holistic approach.

Acknowledgments

This work was part-funded by project LeanBigData : Ultra-Scalable and Ultra-Efficient Integrated and Visual Big Data Analytics (FP7-619606), and by the ERDF -European Regional Development Fund through the Operational Programme for Competitiveness and Internationalisation -COMPETE 2020 Programme within project POCI-01-0145-FEDER-006961 , and by National Funds through the FCT -Fundac ¸ão para a Ciência e a Tecnologia (Portuguese Foundation for Science and Technology) as part of project UID/EEA/50014/2013.