The generation of massive amounts of data in di erent forms (such as activity logs and measurements produced by sensor networks) increased the need of novel data mining algorithms which are capable to build accurate models e ciently and in a distributed fashion. In the recent years, several approaches able to distribute the workload to several machines have been proposed for the classical clustering, classi cation and regression tasks. However, they often su er from strong limitations in their applicability, i.e. they are limited to data organized in a speci c structure, are able to analyze only low-dimensional data, or su er from overhead and scalability issues when the number of instances and attributes increase considerably. For example, to the best of our knowledge, in the literature there are very few density-based clustering algorithms (mostly inspired by the well known algorithm DBSCAN) able to work with an arbitrary number of features and with good scalability performances. Moreover, most of them are exploitable only for pure clustering purposes. At this respect, in this paper we propose a novel density-based clustering algorithm, implemented in the Apache Spark framework, which is able to handle large-scale and high-dimensional data by exploiting the Locality Sensitive Hashing (LSH). Moreover, the proposed approach is able to exploit the identi ed clusters, built on historical data, to forecast the values assumed by a target variable in the future. Therefore, the proposed method can be adopted also for forecasting purposes.
The rest of the paper is structured as follows: in Section 2, we brie y review existing methods to solve the classical density-based clustering task as well as recently proposed clustering methods which are able to handle large-scale data in parallel on multiple processors. In Section 3, we propose our distributed density-based clustering solution, which is able to work on high-dimensional and large-scale data, and describe our approach to exploit the identi ed clusters for forecasting purposes. In Section 4, we report the results of our experimental evaluation, showing that the proposed method is able to obtain accurate predictions and appears very e cient in dealing massive amounts of high-dimensional data. Finally, in Section 5 we draw some conclusions and outline some future work.
Density-based clustering was introduced in [
In order to be able to process large datasets, other works focused on the
computational complexity, which is originally O(n2 m) (where n is the number
of objects and m is the number of features), dominated by the computation of the
neighborhood of each node. An example can be found in [
In this context, the algorithm proposed here attacks the issues raised by
high-dimensional and large-scale datasets through an approach which is
computationally e cient and distributed in all its steps and that, inspired by works on
predictive clustering [
In this Section, we describe our distributed density-based clustering method, which is able to work with high-dimensional and large-scale datasets and that can be exploited for forecasting purposes. In Figure 1 we show the general work ow of the proposed approach, while in the following subsections we describe in detail each step of the method, brie y analyzing their computational complexity and showing that they can be easily parallelized in the Apache Spark framework. 3.1
Given a dataset D of n objects described by m features (m 1 descriptive
attributes and one target attribute), the rst goal is to identify groups of similar
objects according to their features. Most clustering algorithms, including
densitybased algorithms, strongly rely on the computation of similarity/distance
measures among pairs of objects. This task has an inherent cost of O(n2 m), which
can be reduced signi cantly only by resorting to approximated strategies. In this
paper, inspired by [
Once the neighborhood graph has been computed, we identify the set of core objects, according to the de nitions provided in the previous section, that is, we identify the set cores = fo s.t. jN (o)j minP tsg. Let E be the set of edges of the neighborhood graph identi ed by LSH, represented as a distributed data structure of pairs of nodes. This step can be implemented as an aggregateByKey operation (to nd the set of neighbors for each node) followed by a lter operation (to select only those nodes o such that jN (o)j minP ts). Therefore, this step is fully parallelizable and, since each edge of the neighborhood graph is analyzed once, its computational complexity is O(jEj).
Fig. 1. Work ow of the proposed method. 3.2
Given the set of core nodes and the neighborhood graph, our density-based clustering algorithm exploits the property of density-connection, described in Section 2, to simultaneously identify multiple density-based clusters. It is noteworthy that in this phase we do not need node attributes, therefore we can distribute over the available machines only the neighborhood graph, making the proposed algorithm e cient also in terms of memory requirements. An intuitive pseudo-code description of our clustering method is shown in Algorithm 1.
In detail, the algorithm initially assigns a di erent cluster ID to each core node, and the I D = 0 to boundary and noise nodes (lines 1-10). This step can be implemented in a distributed fashion by a single map operation over the set of nodes, therefore its complexity is O(n). Subsequently, each core node sends a message containing its cluster ID to its neighbouring nodes having a lower cluster ID (lines 15-27). Each node receives multiple messages from its adjacent Algorithm 1 Density-based clustering exploiting the neighborhood graph. Require:
G = hV; Ei: the neighborhood graph, where V are the nodes and E are the edges representing the neighborhood relationship between pairs of nodes; cores: set of core nodes from which to start the propagation of cluster IDs; labelChangeRate: minimum percentage of label changes to perform a new iteration. Ensure:
the updated graph G, where nodes V are associated to their cluster ID. fEach node receives multiple messages from (core, clustered) neighboring nodesg for all e = hsrc; dsti 2 E do if src 2 cores and src:clusterID > dst:clusterID then dst:messages dst:messages [ fsrc:clusterIDg changes changes + 1 end if end for nodes and aggregates them by taking the highest cluster ID (lines 29-32)2. We repeat this process until the cluster assignments appear stable, i.e. when the messages exchanged among nodes through the edges of the neighborhood graph are less then a given percentage labelChangeRate of the whole set of edges. This process is coherent with the general concepts at the basis of classical densitybased clustering approaches, since clusters can only be built from core nodes and since the propagation of cluster IDs is performed only from core nodes.
This iterative process can be easily parallelized, since each edge of the graph
can be analyzed independently of the other to evaluate whether a message has
to be sent. Moreover, the nal aggregation step of the received messages can be
performed by a single reduce operation. Therefore, the computational cost of this
phase is dominated by O(u jEj), where u is the number of performed iterations.
We also emphasize the fact that, contrary to existing methods (e.g., [
Inspired by predictive clustering approaches [
Fig. 2. Outline of the forecasting step. comparing their (m 1)-dimensional feature vector (excluding the target attribute, since it is unknown for unlabeled instances) to all the representative vectors associated to the clusters, in order to nd the closest cluster. The value assumed by the target attribute of the closest representative vector is nally associated to the unlabeled instance. Also this step can be easily parallelized through a map operation over the set of unlabeled instances which compares their vector to all the k representative vectors. Therefore, the computational complexity of the prediction phase is O(k m) for each unlabeled instance. 4
We performed the experiments by adopting the self-adaptive online training
strategy, where data are represented as a time-based sliding window [
The considered approaches, each run in its best con guration, are:
Our method, optimized by means of a grid search on a separate set of days, in order to identify the best values of the parameters eps and minP ts;
K-Means algorithm, implemented in the Apache Spark framework, extended with our forecasting strategy. A grid search has been performed using a separate set of days, in order to identify the best value of the parameter k;
ARIMA [
AVG, a baseline approach which predicts the value of the target attribute as the average value assumed over the training instances.
The considered datasets are described in the following:
PVItaly. This dataset contains data about the energy production of photovoltaic power plants, aggregated hourly, collected at regular intervals by sensors located on 17 plants in Italy, from January 1st 2012 to May 4th 2014.
PVNREL. This dataset originally consists of simulated photovoltaic data of 6,000 plants, aggregated hourly, for the year 2006. The experiments were performed on a reduced version of the dataset, consisting of 48 plants belonging to the 16 States with the highest Global Horizontal Irradiation.
Bike Sharing (BS). This dataset4 contains data of the Capital bikeshare
system, aggregated hourly, about the process of rental and returning of bikes,
from/to possibly di erent positions, collected in 2011 and 2012 [
In the results shown in Table 2, we can observe that the simple baseline approach AVG actually appears comparable to ARIMA in the datasets PVNREL and BS. However, they are strongly outperformed by our method and the extended version of K-Means. Our density-based method leads to the best results in PVItaly and PVNREL, whereas it appears comparable to K-Means in BS. Moreover, it can be observed that, on PVNREL and BS, the proposed method is able to exploit the possible availability of more instances in the training phase. At this respect, in order to evaluate whether we can handle a massive amount of data, without incurring in time complexity issues, we performed a scalability test with the large version of the dataset PVNREL, on a cluster of 4 worker machines and one driver machine, each with 4 cores and 32GB of RAM, by progressively increasing the number of edges in the neighborhood graph up to 130 millions. In Figure 3, we can observe that i) the proposed algorithm scales linearly with respect to the number of edges, therefore it is able to deal massive amounts of data, and ii) the possible overhead introduced by the distribution of data to (and by the collection of the results from) di erent machines appears negligible.
In this paper, we proposed a distributed density-based clustering approach, which can be fruitfully exploited for forecasting purposes. The algorithm, implemented in the Apache Spark framework, is fully parallel, does not require any nal merging procedure and is computationally e cient. Experimental results show that it is able to obtain good predictive performance and that it scales linearly with respect to the number of edges in the neighborhood graph. For future work, we intend to extend the forecasting capabilities to the multi-target setting and to perform more extensive comparisons with existing parallel solutions.
We would like to acknowledge the support of the European Commission through the projects MAESTRA - Learning from Massive, Incompletely annotated, and Structured Data (Grant Number ICT-2013-612944) and TOREADOR - Trustworthy Model-aware Analytics Data Platform (Grant Number H2020-688797).