Algorithms for computing All-Pairs Shortest-Paths (APSP) are critical building blocks underlying many practical applications. The standard sequential algorithms, such as Floyd-Warshall and Johnson, quickly become infeasible for large input graphs, necessitating parallel approaches. In this work, we provide detailed analysis of parallel APSP performance on distributed memory clusters with Apache Spark. The Spark model allows for a portable and easy to deploy distributed implementation, and hence is attractive from the end-user point of view. We propose four different APSP implementations for large undirected weighted graphs, which differ in complexity and degree of reliance on techniques outside of pure Spark API. We demonstrate that Spark is able to handle APSP problems with over 200,000 vertices on a 1024-core cluster, and can compete with a naive MPI-based solution. However, our best performing solver requires auxiliary shared persistent storage, and is over two times slower than optimized MPI-based solver.
@cite_11 , propose recursive 2D block-cyclic algorithm that achieves lower-bound on communication latency and bandwidth, which they next extend to 2.5D communication avoiding formulation. On machine with 24,576 cores, the algorithms maintain strong scaling for problems with @math nodes, and weak scaling for up to @math nodes. The paper is also noteworthy for its extensive review of distributed memory APSP solvers. The advantage of the recursive formulation is induced data locality, which directly contributes to improved performance @cite_11 . However, in Spark, the concept of data locality is much weaker, since Spark's runtime system has a significant freedom in scheduling where to materialize or move data for computations. Even though programmer has some control over the data placement (e.g., via partitioning functions), it does not have direct control over where computations will be executed. Therefore, in our work, we derive from the iterative formulation of the 2D block-cyclic algorithm, which can be found in @cite_24 , and is one of several methods tracing their origin to transitive closure @cite_8 .
@cite_24 @cite_8 @cite_11
2107651134 201315547 1986834688 2085399064
We consider distributed memory algorithms for the all-pairs shortest paths (APSP) problem. Scaling the APSP problem to high concurrencies requires both minimizing inter-processor communication as well as maximizing temporal data locality. The 2.5D APSP algorithm, which is based on the divide-and-conquer paradigm, satisfies both of these requirements: it can utilize any extra available memory to perform asymptotically less communication, and it is rich in semiring matrix multiplications, which have high temporal locality. We start by introducing a block-cyclic 2D (minimal memory) APSP algorithm. With a careful choice of block-size, this algorithm achieves known communication lower-bounds for latency and bandwidth. We extend this 2D block-cyclic algorithm to a 2.5D algorithm, which can use c extra copies of data to reduce the bandwidth cost by a factor of c1 2, compared to its 2D counterpart. However, the 2.5D algorithm increases the latency cost by c1 2. We provide a tighter lower bound on latency, which dictates that the latency overhead is necessary to reduce bandwidth along the critical path of execution. Our implementation achieves impressive performance and scaling to 24,576 cores of a Cray XE6 supercomputer by utilizing well-tuned intra-node kernels within the distributed memory algorithm. Extra memory allows parallel matrix multiplication to be done with asymptotically less communication than Cannon's algorithm and be faster in practice. "3D" algorithms arrange the p processors in a 3D array, and store redundant copies of the matrices on each of p1 3 layers. "2D" algorithms such as Cannon's algorithm store a single copy of the matrices on a 2D array of processors. We generalize these 2D and 3D algorithms by introducing a new class of "2.5D algorithms". For matrix multiplication, we can take advantage of any amount of extra memory to store c copies of the data, for any c ∈ 1, 2,..., ⌊p1 3⌋ , to reduce the bandwidth cost of Cannon's algorithm by a factor of c1 2 and the latency cost by a factor c3 2. We also show that these costs reach the lower bounds, modulo polylog(p) factors. We introduce a novel algorithm for 2.5D LU decomposition. To the best of our knowledge, this LU algorithm is the first to minimize communication along the critical path of execution in the 3D case. Our 2.5D LU algorithm uses communicationavoiding pivoting, a stable alternative to partial-pivoting. We prove a novel lower bound on the latency cost of 2.5D and 3D LU factorization, showing that while c copies of the data can also reduce the bandwidth by a factor of c1 2, the latency must increase by a factor of c1 2, so that the 2D LU algorithm (c = 1) in fact minimizes latency. We provide implementations and performance results for 2D and 2.5D versions of all the new algorithms. Our results demonstrate that 2.5D matrix multiplication and LU algorithms strongly scale more efficiently than 2D algorithms. Each of our 2.5D algorithms performs over 2X faster than the corresponding 2D algorithm for certain problem sizes on 65,536 cores of a BG P supercomputer. With the emergence of thread-level parallelism as the primary means for continued performance improvement, the programmability issue has reemerged as an obstacle to the use of architectural advances. We argue that evolving legacy libraries for dense and banded linear algebra is not a viable solution due to constraints imposed by early design decisions. We propose a philosophy of abstraction and separation of concerns that provides a promising solution in this problem domain. The first abstraction, FLASH, allows algorithms to express computation with matrices consisting of contiguous blocks, facilitating algorithms-by-blocks. Operand descriptions are registered for a particular operation a priori by the library implementor. A runtime system, SuperMatrix, uses this information to identify data dependencies between suboperations, allowing them to be scheduled to threads out-of-order and executed in parallel. But not all classical algorithms in linear algebra lend themselves to conversion to algorithms-by-blocks. We show how our recently proposed LU factorization with incremental pivoting and a closely related algorithm-by-blocks for the QR factorization, both originally designed for out-of-core computation, overcome this difficulty. Anecdotal evidence regarding the development of routines with a core functionality demonstrates how the methodology supports high productivity while experimental results suggest that high performance is abundantly achievable. Parallel communication algorithms and networks are central to large-scale parallel computing and, also, data communications. This paper identifies adverse source-destination traffic patterns and proposes a scheme for obtaining relief by means of randomized routing of packets on simple extensions of the well-known omega networks. Valiant and Aleliunas have demonstrated randomized algorithms, for a certain context which we call nonrenewal, that complete the communication task in time O (log N ) with overwhelming probability, where N is the number of sources and destinations. Our scheme has advantages because it uses switches of fixed degree, requires no scheduling, and, for the nonrenewal context, is as good in proven performance. The main advantage of our scheme comes when we consider the renewal context in which packets are generated at the sources continually and asynchronously. Our algorithm extends naturally from the nonrenewal context. In the analysis in the renewal context we, first, explicitly identify the maximum traffic intensities in the internal links of the extended omega networks over all source-destination traffic specifications that satisfy loose bounds. Second, the benefits of randomization on the stability of the network are identified. Third, exact results, for certain restricted models for sources and transmission, and approximate analytic results, for quite general models, are derived for the mean delays. These results show that, in the stable regime, the maximum mean time from source to destination is asymptotically proportional to log N . Numerical results are presented.