# A Distributed Execution Pipeline for Clustering Trajectories Based on a Fuzzy Similarity Relation
## 4 Materials and Methods
### The Overall Execution Pipeline  
The activity in Figure 3 reflects the overall process. All the actions in the diagram are distributed. First, we leverage Spark's _wholeTextFiles_ operation to read the different directories. Also, we specify the number of partitions. While experimenting, we observed that if we specify n partitions we get 2*n ones. After reading the files, we use the Spark's _map_ operation to retrieve the trajectory from each file. Spark's _map_ function is a higher order function taking as a parameter another function which gets specified by us. Figure 4 illustrates the trajectories' creation process. Each file is handled in a distributed manner; however, each line of the files is handled sequentially to create instances of the _PointEntry_ class. Afterward, we evaluated the system and we observed a relatively large time overhead. This led us to specify our own custom partitioning strategy based on the trajectories' sizes. The strategy is implemented in the _LoadPartitioner_ class which extends Spark's _Partitioner_ abstract class. The partitioner has to override the functions _numPartitions_ and _getPartition_ which return the id of the chosen partition. Details are in Algorithm 6, the main idea behind the partitioning strategy is to return the partition with the lowest size. The size of the partitions is not defined by the number of the trajectories, but the size of the points in the trajectories. The efficiency of the logic is provided in Section 5. Figure 4: The creation of the trajectories. Figure 3: The execution pipeline main activity diagram. ```
begin  class LoadPartitioner(partitions:Int) extends Partitioner {  val map:HashMap[Int,Int]=HashMap() // partition, size Functiondef initializeMap  for(i<-0 until partitions)map.put(i,0)  def numPartitions: Int = partitions Functiondef getPartition(key:Any):Int=  key match{ case size:Int=>{ ifmap.isEmptythen initializeMap  val min_elt=map.minBy(__2) // get the lowest size partition  map.put(min_elt_1,min_elt_2+size) // increment its size  min_elt_1 // return it  } }
```  
**Algorithm 6**The _LoadPartitioner_ class. Now that the partitions got repartitioned successfully, we zip each trajectory with a unique identifier. Then, we suppress the trajectories's sizes and conduct the cartesian product to assemble the similarities' elements. After the cartesian product, we conduct a special filter. This filter enables us to reduce the load to more than the half of the overall couples. Instead of computing the similarities of \(n^{2}\) elements, we compute only \(A_{n}^{1}+A_{n-1}^{1}+...+A_{1}^{1}-n\). This is achieved by considering only the couples where the first trajectory's id is inferior than the second. Because the relation is a fuzzy proximity relation, we do only have to compute \(R(x,y)\) not also \(R(y,x)\). Also, we do not have to consider \(R(x,x)\) because it equals 1. Next, we compute the similarities and conduct a filter to suppress the similarities with a weight inferior than the chosen \(\alpha\)-level. The importance of repartitioning the similarities after the filter is proven in the Section 5. After repartitioning, we construct the _SiMatrix_ object and compute the closure. For the clustering, we consider the max-min transitivity and extract the clusters by defining the connected elements of the created Graph object belonging to the Spark's GraphX library. All the extracted clusters form maximum cliques. Now that the clusters are identified, we index each element by its relative cluster, collect and broadcast the identifiers in a Map type form. The rationale behind this broadcast is to reduce the time overhead for the join. Because Spark's joins require shuffle of the data and this can cost a lot of time. However, the broadcast enables us to cache the identifiers in each worker. The join and persistence of the objects are explained in the Figure 5.  
Instead of handling each element, we handle each partition to reduce the number of database connections. For persisting the results, we leverage the Reactive MongoDB Driver.