In all subsequent experiments Aspen was built according to the specs of its developers (compiled with g++-7 -fcilkplus with Cilk Plus runtime libraries) and given access to all 48 cores and full memory on the new lab machine.

On the kron_13_fourth stream:

Ran against naive Kruskal's, NetworkX and our implementation of connected components on graphs with density n^2*0.95 Time is measured in wall-clock seconds.

*: Dubious, since the program did not terminate on its own, but still gave output that signaled the end of execution.

Values in tables are numbers averaged over 10 trials (different graphs). In each experiment we profile ingestion and algorithm execution (this is necessary given the way the bipartiteness test reduction to CC must be implemented). For random graph generation: the probability of an edge appearing was set to 0.03, the geometric insertion/removal parameter was set to 0.5, and there is no limit to how many times an edge may appear in the stream. Experiments were conducted on the lab machine.

Each entry in table is average peak number of bytes. Data recorded using valgrind tool massif with the flags --depth=1 and --ignore-fn=main enabled.

Performance numbers using pthread for the graph workers and without any sketch level parallelism.
Runtime given is seconds (s) or minutes (m). Number within brackets is average updates per second (k is thousand, m is million).

These numbers are from raw ingestion into the buffer tree without any sort of graph streaming work being done

Testing the cluster on a simple word count Apache Spark program.
Input size: 600MB

4 worker nodes:
--- 166.97850894927979 seconds ---
--- 163.22055530548096 seconds ---
--- 164.37560272216797 seconds ---
--- 171.68829035758972 seconds ---

3 worker nodes:
--- 187.09018564224243 seconds ---
--- 184.9761152267456 seconds ---
--- 188.4610195159912 seconds ---

2 worker nodes:
--- 361.66504549980164 seconds ---
--- 362.92039012908936 seconds ---
--- 361.94735503196716 seconds ---

1 worker nodes:
--- 1067.6949362754822 seconds ---
--- 1059.5218110084534 seconds ---
--- 1053.6695115566254 seconds ---

0 node:
--- 2056.108319759369 seconds ---
--- 2004.2446208000183 seconds ---
--- 2048.037629365921 seconds ---

34 worker nodes: ---
--- 84.35914134979248 seconds --- ---
--- 82.6325478553772 seconds --- ---
--- 80.98088264465332 seconds --- ---

GraphX by default works for Scala only. There is a Python wrapper for GraphFrames which is what I'm using. I was able to run their connected components algorithm on the facebook friends graph sample that they have, but this was done locally (not on the cluster).

1. Use spark-submit instead of pyspark so I can run it on the cluster. (Could use advice from Tyler but can do this myself if he's busy)
2. Set up a Hadoop file system on the cluster. (Need some advice from Tyler)
3. Modify the ansible playbook so the workers can downgrade scala/spark version themselves. Almost done with this.
4. Figure out how to load our own input graphs into the algorithm. (Evan might have some suggestions on this).
5. Get the final datasets from Ahmed onto the cluster and test GraphX on it. How big are the datasets? The master node of the cluster only has 350GB of storage and each worker has 30GB. I do not know if HDFS allows us to "spread" a large dataset across all the nodes.
6. Make it so we can vary the number of nodes. There should be a way to do this efficiently/elegantly with spark-submit, but if not we can just use ansible.

16 GB RAM restriction experiments using Tyler's read buffering (buffer size 10^6) and batch size 10^6.

Given the latest sketch update speed test results per Kenny, we can now estimate the input sizes at which our data structure becomes smaller than that of various alternatives (Boost, Networkx). We can also estimate how long our algorithm will take to run on such input sizes, giving us a sense of how close we are to outperforming the current state of the art for in-memory computation.

The above plot displays the space curves for 3 competitors (Networkx's edge list data structure, and Boost's adjacency list and adjacency matrix data structures). All 3 have O(n^2) space complexity but the constants differ. We also plot space curves for 3 variants of our sketching data structure: the currently implemented version which uses 16B per bucket, a trimmed version which reduces the size of error-checking variable c resulting in 12B per bucket, and a further trimmed version which uses half the typical number of buckets (reducing the number of repetitions used to avoid failure whp) for an effective 6B per bucket. We denote the input sizes at which each of our sketching data structures requires no more memory than the most compact alternative (Boost adjacency matrix), and note the data structure sizes of these break-even points.

We can now estimate how long it will take for our algorithm to run on the input sizes specified by these break-even points. We can see that the ~62% reduction in space from our current implementation to my proposed trimmed-bucket implementation makes an enormous difference in the length of time required to "beat" the Boost adjacency matrix algorithm - 30 hours as opposed to hundreds. This suggests that space savings are crucially valuable and that we are nearing the point at which several more iterations of runtime optimization (vectorizing sketch updates, distributing work via Spark, multithreading, buffering via WODS) are all that is required to make our algorithm feasible on inputs that alternative tools cannot match.

Further areas to consider:

• What do we need to know to make similar comparisons for existing external-memory graph solutions?
• How does our algorithm work on more "typical" inputs (larger node set, much sparser graphs) that are present in graph benchmarks such as the GAP benchmark suite? It's likely that on such graphs, we need a lot of space (which we hope can simply be on disk) but suffer little to no penalty in stream processing time per edge (and since the graphs are sparse, the edge stream should be much shorter). Ex: On GAP's Kron benchmark graph with 135M nodes and 2B edges, we only expect about a 5x slowdown in sketch update speed compared to a graph on 100K nodes (assuming that l_0 sketching on longer vectors doesn't matter much - Kenny will verify). However, our data structure size for this graph will be about 60 TB, forcing us to store this structure on disk and use WODS techniques to make sure that stream ingestion is I/O efficient. However, each sketch will only be about 0.5 MB, so they are still easy to distribute.

We need to decide which distributed framework to target since there may be issues with us using Spark. Add details about different options (including what we know about Spark) here.

Notes on integrating C++ code into PySpark https://www.slideshare.net/databricks/integrating-existing-c-libraries-into-pyspark-with-esther-kundin

• People have done it before
• This guide might be helpful
• There are some performance considerations - they recommend not keeping state in the C++ code? Am I misinterpreting that?
• seems like a fair bit of added complexity. Though I'm not sure how much.

Another tutorial on using C++ with Spark. At a quick glance this one seems simpler. https://medium.com/@thomaspt748/integrate-c-c-libraries-dll-so-into-apache-spark-scala-in-hadoop-cluster-56cdb26e15ea

Thrill - a C++ native distributed big data framework https://github.com/thrill/thrill

• C++ native should make integration much easier
• far less used than Spark, not an industry standard
• EXPERIMENTAL

Apache Spark

• Built on top of Hadoop, requires knowledge and understanding of Hadoop
• Widely used in industry and lots of functionality/features
• Only available in Java/Scala/Python

Open MPI

• Available in C/C++
• API forces user to handle inter-process communication

Map Reduce for C/C++ framework on Hadoop

• GitHub repo for the code: https://github.com/google/mr4c
• Article written about it: https://gigaom.com/2015/02/18/google-open-sources-a-mapreduce-framework-for-c/
• It hasn't been updated for awhile. Could be complete, could be deprecated.
• Spark is known to run faster than MapReduce?

Naive impl: insert, periodically query all updates for each sketch, batch all of them, delete from DS

• Put
• Get
• Delete

Better impl: Insert and modify flushing algorithm to batch updates given some heurstic (e.g. all updates are for same sketch, depth of tree, etc)

• Put
• Low level access to flush (e.g. custom flush, flush hook)

Given a buffer tree with B^e = 16, buffer size 4096, n = 10^10 for naive, n = 10^5 for better, naive runs roughly 10-100x slower.

Each node in the above tree represents a fully external memory (EM) graph processing model (GPM) in which connected components (CC) is implemented. Each directed edge indicates that the destination node GPM demonstrates a CC algorithm which is faster than the CC algorithm implemented in the source node GPM. The requirement for such a demonstration is an experiment comparing the performance of the two algorithms across graphs varying in scale on identical hardware in the same environment. The sources to the studies supporting each edge is provided below in the references section.

• Although X-Stream is technically not a fully EM GPM, it is included because it is used as a performance milestone for many other fully EM GPMs.
• NXGraph demonstrates it is faster than TurboGraph for real world graphs at most the scale of the Twitter graph (1.47 billion edges), however NXGraph runtime eventually overtakes TurboGraph for larger graphs. This performance superiority on different sized graphs is indicated by the two directed edges between NXGraph and TurboGraph.
• NXGraph claims speedups over GridGraph and MMap, however these experiments only run a single iteration of PageRank for a single graph.
• Mosaic and Graphene are the first fully EM GPMs which support computations on graphs with 1 trillion edges.
• MOSAIC is the first major GPM to begin exploiting NVMe architecture. GraFBoost follows suit, while CLIP extends optional leveraging of NVMe.
• Graphene excludes preprocessing time from its performance comparison, simply claiming that competitor preprocessing time is "similar or longer"
• CLIP compares itself to MOSAIC on a commodity PC, without the massively-parallel Xeon Phi processors MOSAIC was intended to leverage for its graph computation. The MOSAIC paper contains CPU-only tests which still prove it is superior to GridGraph, however it should be understood that CLIP has only been shown to be superior this CPU-only version of MOSAIC, and that for massive graphs (>256 GB representation size) the full MOSAIC framework may or may not outperform CLIP.
• Though LUMOS does not explicitly implement a CC algorithm, the original paper proves that it performs label propagation faster than GridGraph. Label Propagation is the basis for most CC algorithms implemented by GPMs, including GridGraph.
• Similarly, V-Part compares BFS instead of CC against GraFBoost, however these similar "traversal" algorithms are essentially just different versions of the aforementioned label propagation and thus exhibit similar runtimes across GPMs.

Below are the graph data sets used by some of the most popular GPMs. Most other GPMs evaluate themselves on some subset of these graphs as well.

The netflix graph is bipartite.

Note: rmat data sets are synthetic graphs.

The experimental evidence supporting (almost) each edge in the graph may be found in the original publication of its destination node GPM.

• GraphChi
• X-Stream
• TurboGraph
• LigraChi-g
• MMap
• PrefEdge
• BPP
• GridGraph
• GPSA
• PathGraph
• DS-GraphChi (GraphChi with dynamic shards)
• VENUS
• NXGraph
• AsyncStripe
• Graphene
• MOSAIC
• GraFBoost (Soft)
• Wonderland
• CLIP
• V-Part
• LUMOS

• The edge from DS-GraphChi to X-Stream is supported by the DS-GraphChi paper.
• The edge from LigraChi-g to GridGraph is supported by the Wonderland paper.

The above "imporvement tree" is incomplete. The LaTeX source code used to generate the above improvement tree using the tikz package is provided here for inclusion in other documents or expansion of the tree to include more GPMs and experiments.

% Node style
\tikzstyle{textcontainer}=[fill=none, draw=black, shape=rectangle]
 
% Edge style
\tikzstyle{diredge}=[->]


\begin{tikzpicture}
	\begin{pgfonlayer}{nodelayer}
		\node [style=textcontainer] (0) at (0, 0) {GraphChi};
		\node [style=textcontainer] (1) at (24, -12) {TurboGraph};
		\node [style=textcontainer] (2) at (0, -12) {X-Stream};
		\node [style=textcontainer] (4) at (-12, -12) {LigraChi-g};
		\node [style=textcontainer] (5) at (24, -18) {MMap};
		\node [style=textcontainer] (6) at (17.25, -4.25) {BPP};
		\node [style=textcontainer] (7) at (-24, -12) {PrefEdge};
		\node [style=textcontainer] (8) at (0, -24) {GridGraph};
		\node [style=textcontainer] (9) at (-12, -24) {GPSA};
		\node [style=textcontainer] (10) at (-24, -24) {PathGraph};
		\node [style=textcontainer] (11) at (-3, -6) {DS-GraphChi};
		\node [style=textcontainer] (12) at (12, -24) {VENUS};
		\node [style=textcontainer] (13) at (12, -12) {NXGraph};
		\node [style=textcontainer] (14) at (12, -36) {AsyncStripe};
		\node [style=textcontainer] (16) at (0, -36) {MOSAIC};
		\node [style=textcontainer] (17) at (-24, -36) {Graphene};
		\node [style=textcontainer] (18) at (24, -24) {GraFBoost (Soft)};
		\node [style=textcontainer] (19) at (-12, -36) {Wonderland};
		\node [style=textcontainer] (20) at (0, -48) {CLIP};
		\node [style=textcontainer] (21) at (24, -30) {V-Part};
		\node [style=textcontainer] (22) at (24, -36) {LUMOS};
	\end{pgfonlayer}
	\begin{pgfonlayer}{edgelayer}
		\draw [style=diredge] (0) to (1);
		\draw [style=diredge] (0) to (2);
		\draw [style=diredge] (0) to (4);
		\draw [style=diredge] (1) to (5);
		\draw [style=diredge] (0) to (6);
		\draw [style=diredge] (0) to (7);
		\draw [style=diredge] (2) to (8);
		\draw [style=diredge] (2) to (9);
		\draw [style=diredge] (2) to (10);
		\draw [style=diredge] (0) to (11);
		\draw [style=diredge] (11) to (2);
		\draw [style=diredge] (2) to (12);
		\draw [style=diredge, bend right=15] (1) to (13);
		\draw [style=diredge, bend right=15] (13) to (1);
		\draw [style=diredge] (0) to (13);
		\draw [style=diredge] (8) to (14);
		\draw [style=diredge] (8) to (16);
		\draw [style=diredge] (8) to (17);
		\draw [style=diredge] (2) to (18);
		\draw [style=diredge] (4) to (8);
		\draw [style=diredge] (8) to (19);
		\draw [style=diredge] (16) to (20);
		\draw [style=diredge] (18) to (21);
		\draw [style=diredge] (8) to (22);
	\end{pgfonlayer}
\end{tikzpicture}

open image in new tab to click around in the graph -- (not working right now for some reason... you can click locally though)

My Source

Specifically the commands I used to monitor the cpu activity on the system were:
perf record -F 99 -a -g -- [command (ie ./experiment or sleep)]
perf script | ./stackcollapse-perf.pl > out.perf-folded
./flamegraph.pl out.perf-folded > perf-kernel.svg

This records activity across the entire system while the command is running. Therefore we can record for a time window. Ie 60 seconds using sleep.