Jialin Dong, Da Zheng, Lin F. Yang, Geroge Karypis
Graph neural networks (GNNs) are powerful tools for learning from graph data and are widely used in various applications such as social network recommendation, fraud detection, and graph search. The graphs in these applications are typically large, usually containing hundreds of millions of nodes. Training GNN models on such large graphs efficiently remains a big challenge. Despite a number of sampling-based methods have been proposed to enable mini-batch training on large graphs, these methods have not been proved to work on truly industry-scale graphs, which require GPUs or mixed-CPU-GPU training. The state-of-the-art sampling-based methods are usually not optimized for these real-world hardware setups, in which data movement between CPUs and GPUs is a bottleneck. To address this issue, we propose Global Neighborhood Sampling that aims at training GNNs on giant graphs specifically for mixed-CPU-GPU training. The algorithm samples a global cache of nodes periodically for all mini-batches and stores them in GPUs. This global cache allows in-GPU importance sampling of mini-batches, which drastically reduces the number of nodes in a mini-batch, especially in the input layer, to reduce data copy between CPU and GPU and mini-batch computation without compromising the training convergence rate or model accuracy. We provide a highly efficient implementation of this method and show that our implementation outperforms an efficient node-wise neighbor sampling baseline by a factor of 2X-4X on giant graphs. It outperforms an efficient implementation of LADIES with small layers by a factor of 2X-14X while achieving much higher accuracy than LADIES.We also theoretically analyze the proposed algorithm and show that with cached node data of a proper size, it enjoys a comparable convergence rate as the underlying node-wise sampling method.
Vern I. Paulsen, Da Zheng
We study tensor products and nuclearity-related properties of the operator system $\mathcal S_n$ generated by the Cuntz isometries. By using the nuclearity of the Cuntz algebra, we can show that $\mathcal{S}_n$ is $C^*$-nuclear, and this implies a dual row contraction version of Ando's theorem characterizing operators of numerical radius 1. On the other hand, without using the nuclearity of the Cuntz algebra, we are still able to show directly this Ando type property of dual row contractions and conclude that $\mathcal{S}_n$ is $C^*$-nuclear, which yields a new proof of the nuclearity of the Cuntz algebras. We prove that the dual operator system of $\mathcal{S}_n$ is completely order isomorphic to an operator subsystem of $M_{n+1}$. Finally, a lifting result concerning Popescu's joint numerical radius is proved via operator system techniques.
Da Zheng, Chao Ma, Minjie Wang, Jinjing Zhou, Qidong Su, Xiang Song, Quan Gan, Zheng Zhang, George Karypis
Graph neural networks (GNN) have shown great success in learning from graph-structured data. They are widely used in various applications, such as recommendation, fraud detection, and search. In these domains, the graphs are typically large, containing hundreds of millions of nodes and several billions of edges. To tackle this challenge, we develop DistDGL, a system for training GNNs in a mini-batch fashion on a cluster of machines. DistDGL is based on the Deep Graph Library (DGL), a popular GNN development framework. DistDGL distributes the graph and its associated data (initial features and embeddings) across the machines and uses this distribution to derive a computational decomposition by following an owner-compute rule. DistDGL follows a synchronous training approach and allows ego-networks forming the mini-batches to include non-local nodes. To minimize the overheads associated with distributed computations, DistDGL uses a high-quality and light-weight min-cut graph partitioning algorithm along with multiple balancing constraints. This allows it to reduce communication overheads and statically balance the computations. It further reduces the communication by replicating halo nodes and by using sparse embedding updates. The combination of these design choices allows DistDGL to train high-quality models while achieving high parallel efficiency and memory scalability. We demonstrate our optimizations on both inductive and transductive GNN models. Our results show that DistDGL achieves linear speedup without compromising model accuracy and requires only 13 seconds to complete a training epoch for a graph with 100 million nodes and 3 billion edges on a cluster with 16 machines. DistDGL is now publicly available as part of DGL:https://github.com/dmlc/dgl/tree/master/python/dgl/distributed.
Da Zheng
In this paper we consider the operator system $\cl{S}_n$ generated by $n$ Cuntz isometries, i.e. the span of the generators of the Cuntz algebra $\cl{O}_n$ together with their adjoints and the identity. We define an operator subsystem $\cl{E}_n\subseteq M_{n+1}$ and then prove that $\cl{S}_n$ is completely order isomorphic to a quotient of $\cl{E}_n$. This result implies a characterization of positive elements in $M_p(\cl{S}_n)$.
Da Zheng, Disa Mhembere, Vince Lyzinski, Joshua Vogelstein, Carey E. Priebe, Randal Burns
Sparse matrix multiplication is traditionally performed in memory and scales to large matrices using the distributed memory of multiple nodes. In contrast, we scale sparse matrix multiplication beyond memory capacity by implementing sparse matrix dense matrix multiplication (SpMM) in a semi-external memory (SEM) fashion; i.e., we keep the sparse matrix on commodity SSDs and dense matrices in memory. Our SEM-SpMM incorporates many in-memory optimizations for large power-law graphs. It outperforms the in-memory implementations of Trilinos and Intel MKL and scales to billion-node graphs, far beyond the limitations of memory. Furthermore, on a single large parallel machine, our SEM-SpMM operates as fast as the distributed implementations of Trilinos using five times as much processing power. We also run our implementation in memory (IM-SpMM) to quantify the overhead of keeping data on SSDs. SEM-SpMM achieves almost 100% performance of IM-SpMM on graphs when the dense matrix has more than four columns; it achieves at least 65% performance of IM-SpMM on all inputs. We apply our SpMM to three important data analysis tasks--PageRank, eigensolving, and non-negative matrix factorization--and show that our SEM implementations significantly advance the state of the art.
Da Zheng, Xiang Song, Chao Ma, Zeyuan Tan, Zihao Ye, Jin Dong, Hao Xiong, Zheng Zhang, George Karypis
Knowledge graphs have emerged as a key abstraction for organizing information in diverse domains and their embeddings are increasingly used to harness their information in various information retrieval and machine learning tasks. However, the ever growing size of knowledge graphs requires computationally efficient algorithms capable of scaling to graphs with millions of nodes and billions of edges. This paper presents DGL-KE, an open-source package to efficiently compute knowledge graph embeddings. DGL-KE introduces various novel optimizations that accelerate training on knowledge graphs with millions of nodes and billions of edges using multi-processing, multi-GPU, and distributed parallelism. These optimizations are designed to increase data locality, reduce communication overhead, overlap computations with memory accesses, and achieve high operation efficiency. Experiments on knowledge graphs consisting of over 86M nodes and 338M edges show that DGL-KE can compute embeddings in 100 minutes on an EC2 instance with 8 GPUs and 30 minutes on an EC2 cluster with 4 machines with 48 cores/machine. These results represent a 2x~5x speedup over the best competing approaches. DGL-KE is available on https://github.com/awslabs/dgl-ke.
Da Zheng, Disa Mhembere, Joshua T. Vogelstein, Carey E. Priebe, Randal Burns
R is one of the most popular programming languages for statistics and machine learning, but the R framework is relatively slow and unable to scale to large datasets. The general approach for speeding up an implementation in R is to implement the algorithms in C or FORTRAN and provide an R wrapper. FlashR takes a different approach: it executes R code in parallel and scales the code beyond memory capacity by utilizing solid-state drives (SSDs) automatically. It provides a small number of generalized operations (GenOps) upon which we reimplement a large number of matrix functions in the R base package. As such, FlashR parallelizes and scales existing R code with little/no modification. To reduce data movement between CPU and SSDs, FlashR evaluates matrix operations lazily, fuses operations at runtime, and uses cache-aware, two-level matrix partitioning. We evaluate FlashR on a variety of machine learning and statistics algorithms on inputs of up to four billion data points. FlashR out-of-core tracks closely the performance of FlashR in-memory. The R code for machine learning algorithms executed in FlashR outperforms the in-memory execution of H2O and Spark MLlib by a factor of 2-10 and outperforms Revolution R Open by more than an order of magnitude.
Da Zheng, Lun Du, Junwei Su, Yuchen Tian, Yuqi Zhu, Jintian Zhang, Lanning Wei, Ningyu Zhang, Huajun Chen
Problem-solving has been a fundamental driver of human progress in numerous domains. With advancements in artificial intelligence, Large Language Models (LLMs) have emerged as powerful tools capable of tackling complex problems across diverse domains. Unlike traditional computational systems, LLMs combine raw computational power with an approximation of human reasoning, allowing them to generate solutions, make inferences, and even leverage external computational tools. However, applying LLMs to real-world problem-solving presents significant challenges, including multi-step reasoning, domain knowledge integration, and result verification. This survey explores the capabilities and limitations of LLMs in complex problem-solving, examining techniques including Chain-of-Thought (CoT) reasoning, knowledge augmentation, and various LLM-based and tool-based verification techniques. Additionally, we highlight domain-specific challenges in various domains, such as software engineering, mathematical reasoning and proving, data analysis and modeling, and scientific research. The paper further discusses the fundamental limitations of the current LLM solutions and the future directions of LLM-based complex problems solving from the perspective of multi-step reasoning, domain knowledge integration and result verification.
Da Zheng, Randal Burns, Joshua Vogelstein, Carey E. Priebe, Alexander S. Szalay
Many eigensolvers such as ARPACK and Anasazi have been developed to compute eigenvalues of a large sparse matrix. These eigensolvers are limited by the capacity of RAM. They run in memory of a single machine for smaller eigenvalue problems and require the distributed memory for larger problems. In contrast, we develop an SSD-based eigensolver framework called FlashEigen, which extends Anasazi eigensolvers to SSDs, to compute eigenvalues of a graph with hundreds of millions or even billions of vertices in a single machine. FlashEigen performs sparse matrix multiplication in a semi-external memory fashion, i.e., we keep the sparse matrix on SSDs and the dense matrix in memory. We store the entire vector subspace on SSDs and reduce I/O to improve performance through caching the most recent dense matrix. Our result shows that FlashEigen is able to achieve 40%-60% performance of its in-memory implementation and has performance comparable to the Anasazi eigensolvers on a machine with 48 CPU cores. Furthermore, it is capable of scaling to a graph with 3.4 billion vertices and 129 billion edges. It takes about four hours to compute eight eigenvalues of the billion-node graph using 120 GB memory.
Da Zheng, Xiang Song, Chengru Yang, Dominique LaSalle, George Karypis
Graph neural networks (GNN) have shown great success in learning from graph-structured data. They are widely used in various applications, such as recommendation, fraud detection, and search. In these domains, the graphs are typically large and heterogeneous, containing many millions or billions of vertices and edges of different types. To tackle this challenge, we develop DistDGLv2, a system that extends DistDGL for training GNNs on massive heterogeneous graphs in a mini-batch fashion, using distributed hybrid CPU/GPU training. DistDGLv2 places graph data in distributed CPU memory and performs mini-batch computation in GPUs. For ease of use, DistDGLv2 adopts API compatible with Deep Graph Library (DGL)'s mini-batch training and heterogeneous graph API, which enables distributed training with almost no code modification. To ensure model accuracy, DistDGLv2 follows a synchronous training approach and allows ego-networks forming mini-batches to include non-local vertices. To ensure data locality and load balancing, DistDGLv2 partitions heterogeneous graphs by using a multi-level partitioning algorithm with min-edge cut and multiple balancing constraints. DistDGLv2 deploys an asynchronous mini-batch generation pipeline that makes computation and data access asynchronous to fully utilize all hardware (CPU, GPU, network, PCIe). The combination allows DistDGLv2 to train high-quality models while achieving high parallel efficiency and memory scalability. We demonstrate DistDGLv2 on various GNN workloads. Our results show that DistDGLv2 achieves 2-3X speedup over DistDGL and 18X speedup over Euler. It takes only 5-10 seconds to complete an epoch on graphs with hundreds of millions of vertices on a cluster with 64 GPUs.
Da Zheng, Randal Burns, Alexander S. Szalay
Solid state disks (SSDs) have advanced to outperform traditional hard drives significantly in both random reads and writes. However, heavy random writes trigger fre- quent garbage collection and decrease the performance of SSDs. In an SSD array, garbage collection of individ- ual SSDs is not synchronized, leading to underutilization of some of the SSDs. We propose a software solution to tackle the unsyn- chronized garbage collection in an SSD array installed in a host bus adaptor (HBA), where individual SSDs are exposed to an operating system. We maintain a long I/O queue for each SSD and flush dirty pages intelligently to fill the long I/O queues so that we hide the performance imbalance among SSDs even when there are few parallel application writes. We further define a policy of select- ing dirty pages to flush and a policy of taking out stale flush requests to reduce the amount of data written to SSDs. We evaluate our solution in a real system. Experi- ments show that our solution fully utilizes all SSDs in an array under random write-heavy workloads. It improves I/O throughput by up to 62% under random workloads of mixed reads and writes when SSDs are under active garbage collection. It causes little extra data writeback and increases the cache hit rate.
Heng Wang, Da Zheng, Randal Burns, Carey Priebe
A canonical problem in graph mining is the detection of dense communities. This problem is exacerbated for a graph with a large order and size -- the number of vertices and edges -- as many community detection algorithms scale poorly. In this work we propose a novel framework for detecting active communities that consist of the most active vertices in massive graphs. The framework is applicable to graphs having billions of vertices and hundreds of billions of edges. Our framework utilizes a parallelizable trimming algorithm based on a locality statistic to filter out inactive vertices, and then clusters the remaining active vertices via spectral decomposition on their similarity matrix. We demonstrate the validity of our method with synthetic Stochastic Block Model graphs, using Adjusted Rand Index as the performance metric. We further demonstrate its practicality and efficiency on a most recent real-world Hyperlink Web graph consisting of over 3.5 billion vertices and 128 billion edges.
Da Zheng, Alexander Szalay, Andreas Terzis
In our previous work we introduced a so-called Amdahl blade microserver that combines a low-power Atom processor, with a GPU and an SSD to provide a balanced and energy-efficient system. Our preliminary results suggested that the sequential I/O of Amdahl blades can be ten times higher than that a cluster of conventional servers with comparable power consumption. In this paper we investigate the performance and energy efficiency of Amdahl blades running Hadoop. Our results show that Amdahl blades are 7.7 times and 3.4 times as energy-efficient as the Open Cloud Consortium cluster for a data-intensive and a compute-intensive application, respectively. The Hadoop Distributed Filesystem has relatively poor performance on Amdahl blades because both disk and network I/O are CPU-heavy operations on Atom processors. We demonstrate three effective techniques to reduce CPU consumption and improve performance. However, even with these improvements, the Atom processor is still the system's bottleneck. We revisit Amdahl's law, and estimate that Amdahl blades need four Atom cores to be well balanced for Hadoop tasks.
Da Zheng, Disa Mhembere, Randal Burns, Joshua Vogelstein, Carey E. Priebe, Alexander S. Szalay
Graph analysis performs many random reads and writes, thus, these workloads are typically performed in memory. Traditionally, analyzing large graphs requires a cluster of machines so the aggregate memory exceeds the graph size. We demonstrate that a multicore server can process graphs with billions of vertices and hundreds of billions of edges, utilizing commodity SSDs with minimal performance loss. We do so by implementing a graph-processing engine on top of a user-space SSD file system designed for high IOPS and extreme parallelism. Our semi-external memory graph engine called FlashGraph stores vertex state in memory and edge lists on SSDs. It hides latency by overlapping computation with I/O. To save I/O bandwidth, FlashGraph only accesses edge lists requested by applications from SSDs; to increase I/O throughput and reduce CPU overhead for I/O, it conservatively merges I/O requests. These designs maximize performance for applications with different I/O characteristics. FlashGraph exposes a general and flexible vertex-centric programming interface that can express a wide variety of graph algorithms and their optimizations. We demonstrate that FlashGraph in semi-external memory performs many algorithms with performance up to 80% of its in-memory implementation and significantly outperforms PowerGraph, a popular distributed in-memory graph engine.
Da Zheng, Xiang Song, Qi Zhu, Jian Zhang, Theodore Vasiloudis, Runjie Ma, Houyu Zhang, Zichen Wang, Soji Adeshina, Israt Nisa, Alejandro Mottini, Qingjun Cui, Huzefa Rangwala, Belinda Zeng, Christos Faloutsos, George Karypis
Graph machine learning (GML) is effective in many business applications. However, making GML easy to use and applicable to industry applications with massive datasets remain challenging. We developed GraphStorm, which provides an end-to-end solution for scalable graph construction, graph model training and inference. GraphStorm has the following desirable properties: (a) Easy to use: it can perform graph construction and model training and inference with just a single command; (b) Expert-friendly: GraphStorm contains many advanced GML modeling techniques to handle complex graph data and improve model performance; (c) Scalable: every component in GraphStorm can operate on graphs with billions of nodes and can scale model training and inference to different hardware without changing any code. GraphStorm has been used and deployed for over a dozen billion-scale industry applications after its release in May 2023. It is open-sourced in Github: https://github.com/awslabs/graphstorm.
Timothy M. Chan, Da Wei Zheng
We revisit Hopcroft's problem and related fundamental problems about geometric range searching. Given $n$ points and $n$ lines in the plane, we show how to count the number of point-line incidence pairs or the number of point-above-line pairs in $O(n^{4/3})$ time, which matches the conjectured lower bound and improves the best previous time bound of $n^{4/3}2^{O(\log^*n)}$ obtained almost 30 years ago by Matoušek. We describe two interesting and different ways to achieve the result: the first is randomized and uses a new 2D version of fractional cascading for arrangements of lines; the second is deterministic and uses decision trees in a manner inspired by the sorting technique of Fredman (1976). The second approach extends to any constant dimension. Many consequences follow from these new ideas: for example, we obtain an $O(n^{4/3})$-time algorithm for line segment intersection counting in the plane, $O(n^{4/3})$-time randomized algorithms for bichromatic closest pair and Euclidean minimum spanning tree in three or four dimensions, and a randomized data structure for halfplane range counting in the plane with $O(n^{4/3})$ preprocessing time and space and $O(n^{1/3})$ query time.
Da Wei Zheng, Monika Henzinger
$\newcommand{\eps}{\varepsilon}$We present an auction algorithm using multiplicative instead of constant weight updates to compute a $(1-\eps)$-approximate maximum weight matching (MWM) in a bipartite graph with $n$ vertices and $m$ edges in time $O(m\eps^{-1})$, beating the running time of the fastest known approximation algorithm of Duan and Pettie [JACM '14] that runs in $O(m\eps^{-1}\log \eps^{-1})$. Our algorithm is very simple and it can be extended to give a dynamic data structure that maintains a $(1-\eps)$-approximate maximum weight matching under (1) one-sided vertex deletions (with incident edges) and (2) one-sided vertex insertions (with incident edges sorted by weight) to the other side. The total time used is $O(m\eps^{-1})$, where $m$ is the sum of the number of initially existing and inserted edges.
Timothy M. Chan, Pingan Cheng, Da Wei Zheng
Polynomial partitioning techniques have recently led to improved geometric data structures for a variety of fundamental problems related to semialgebraic range searching and intersection searching in 3D and higher dimensions (e.g., see [Agarwal, Aronov, Ezra, and Zahl, SoCG 2019; Ezra and Sharir, SoCG 2021; Agarwal, Aronov, Ezra, Katz, and Sharir, SoCG 2022]). They have also led to improved algorithms for offline versions of semialgebraic range searching in 2D, via lens-cutting [Sharir and Zahl (2017)]. In this paper, we show that these techniques can yield new data structures for a number of other 2D problems even for online queries: 1. Semialgebraic range stabbing. We present a data structure for $n$ semialgebraic ranges in 2D of constant description complexity with $O(n^{3/2+\varepsilon})$ preprocessing time and space, so that we can count the number of ranges containing a query point in $O(n^{1/4+\varepsilon})$ time, for an arbitrarily small constant $\varepsilon>0$. 2. Ray shooting amid algebraic arcs. We present a data structure for $n$ algebraic arcs in 2D of constant description complexity with $O(n^{3/2+\varepsilon})$ preprocessing time and space, so that we can find the first arc hit by a query (straight-line) ray in $O(n^{1/4+\varepsilon})$ time. 3. Intersection counting amid algebraic arcs. We present a data structure for $n$ algebraic arcs in 2D of constant description complexity with $O(n^{3/2+\varepsilon})$ preprocessing time and space, so that we can count the number of intersection points with a query algebraic arc of constant description complexity in $O(n^{1/2+\varepsilon})$ time. In particular, this implies an $O(n^{3/2+\varepsilon})$-time algorithm for counting intersections between two sets of $n$ algebraic arcs in 2D.
Timothy M. Chan, Pingan Cheng, Da Wei Zheng
We present the first optimal randomized algorithm for constructing the order-$k$ Voronoi diagram of $n$ points in two dimensions. The expected running time is $O(n\log n + nk)$, which improves the previous, two-decades-old result of Ramos (SoCG'99) by a $2^{O(\log^*k)}$ factor. To obtain our result, we (i) use a recent decision-tree technique of Chan and Zheng (SODA'22) in combination with Ramos's cutting construction, to reduce the problem to verifying an order-$k$ Voronoi diagram, and (ii) solve the verification problem by a new divide-and-conquer algorithm using planar-graph separators. We also describe a deterministic algorithm for constructing the $k$-level of $n$ lines in two dimensions in $O(n\log n + nk^{1/3})$ time, and constructing the $k$-level of $n$ planes in three dimensions in $O(n\log n + nk^{3/2})$ time. These time bounds (ignoring the $n\log n$ term) match the current best upper bounds on the combinatorial complexity of the $k$-level. Previously, the same time bound in two dimensions was obtained by Chan (1999) but with randomization.
Adam Karczmarz, Da Wei Zheng
Le and Wulff-Nilsen [SODA '24] initiated a systematic study of VC set systems to unweighted $K_h$-minor-free directed graphs. We extend their results in the following ways: $\bullet$ We present the first application of VC set systems for real-weighted minor-free digraphs to build the first exact subquadratic-space distance oracle with $O(\log n)$ query time. Prior work using VC set systems only applied in unweighted and integer weighted digraphs. $\bullet$ We describe a unified system for analyzing the VC dimension of balls and the LP set system (based on Li--Parter [STOC '19]) of Le--Wulff-Nilsen [SODA '24] using pseudodimension. This is a major conceptual contribution that allows for both improving our understanding of set systems in digraphs as well as improving the bound of the LP set system in directed graphs to $h-1$. $\bullet$ We present the first application of these set systems in a dynamic setting. Specifically, we construct decremental reachability oracles with subquadratic total update time and constant query time. Prior to this work, it was not known if this was possible to construct oracles with subquadratic total update time and polylogarithmic query time, even in planar digraphs. $\bullet$ We describe subquadratic time algorithms for unweighted digraphs including (1) constructions of exact distance oracles, (2) computation of vertex eccentricities and Wiener index. The main innovation in obtaining these results is the use of dynamic string data structures.