differential-dataflow

Differential Dataflow An implementation of differential dataflow over timely dataflow on Rust. Background Differential dataflow is a data-parallel programming framework designed to efficiently process large volumes of data and to quickly respond to arbitrary changes in input collections. You can read more in the differential dataflow mdbook and in the differential dataflow documentation. Differential dataflow programs are written as functional transformations of collections of data, using familiar operators like map, filter, join, and reduce. Differential dataflow also includes more exotic operators such as iterate, which repeatedly applies a differential dataflow fragment to a collection. The programs are compiled down to timely dataflow computations. For example, here is a differential dataflow fragment to compute the out-degree distribution of a directed graph (for each degree, the number of nodes with that many outgoing edges): let out_degr_dist = edges.map(|(src, _dst)| src) // extract source .count() // count occurrences of source .map(|(_src, deg)| deg) // extract degree .count(); // count occurrences of degree Alternately, here is a fragment that computes the set of nodes reachable from a set roots of starting nodes: let reachable = roots.iterate(|scope, reach| edges.enter(scope) .semijoin(reach) .map(|(src, dst)| dst) .concat(reach) .distinct() ) Once written, a differential dataflow responds to arbitrary changes to its initially empty input collections, reporting the corresponding changes to each of its output collections. Differential dataflow can react quickly because it only acts where changes in collections occur, and does no work elsewhere. In the examples above, we can add to and remove from edges, dynamically altering the graph, and get immediate feedback on how the results change: if the degree distribution shifts we'll see the changes, and if nodes are now (or no longer) reachable we'll hear about that too. We could also add to and remove from roots, more fundamentally altering the reachability query itself. Be sure to check out the differential dataflow documentation, which is continually improving. An example: counting degrees in a graph. Let's check out that out-degree distribution computation, to get a sense for how differential dataflow actually works. This example is examples/hello.rs in this repository, if you'd like to follow along. A graph is a collection of pairs (Node, Node), and one standard analysis is to determine the number of times each Node occurs in the first position, its "degree". The number of nodes with each degree is a helpful graph statistic. To determine the out-degree distribution, we create a new timely dataflow scope in which we describe our computation and how we plan to interact with it. // create a degree counting differential dataflow let (mut input, probe) = worker.dataflow(|scope| { // create edge input, count a few ways. let (input, edges) = scope.new_collection(); let out_degr_distr = edges.map(|(src, _dst)| src) // extract source .count() // count occurrences of source .map(|(_src, deg)| deg) // extract degree .count(); // count occurrences of degree // show us something about the collection, notice when done. let probe = out_degr_distr .inspect(|x| println!("observed: {:?}", x)) .probe(); (input, probe) }); The input and probe we return are how we get data into the dataflow, and how we notice when some amount of computation is complete. These are timely dataflow idioms, and we won't get in to them in more detail here (check out the timely dataflow repository). If we feed this computation with some random graph data, say fifty random edges among ten nodes, we get output like Echidnatron% cargo run --release --example hello -- 10 50 1 inspect Finished release [optimized + debuginfo] target(s) in 0.05s Running `target/release/examples/hello 10 50 1 inspect` observed: ((3, 1), 0, 1) observed: ((4, 2), 0, 1) observed: ((5, 4), 0, 1) observed: ((6, 2), 0, 1) observed: ((7, 1), 0, 1) round 0 finished after 772.464µs (loading) This shows us the records that passed the inspect operator, revealing the contents of the collection: there are five distinct degrees, three through seven. The records have the form ((degree, count), time, delta) where the time field says this is the first round of data, and the delta field tells us that each record is coming into existence. If the corresponding record were departing the collection, it would be a negative number. Let's update the input by removing one edge and adding a new random edge: observed: ((2, 1), 1, 1) observed: ((3, 1), 1, -1) observed: ((7, 1), 1, -1) observed: ((8, 1), 1, 1) round 1 finished after 149.701µs We see here some changes! Those degree three and seven nodes have been replaced by degree two and eight nodes; looks like one node lost an edge and gave it to the other! How about a few more changes? round 2 finished after 127.444µs round 3 finished after 100.628µs round 4 finished after 130.609µs observed: ((5, 3), 5, 1) observed: ((5, 4), 5, -1) observed: ((6, 2), 5, -1) observed: ((6, 3), 5, 1) observed: ((7, 1), 5, 1) observed: ((8, 1), 5, -1) round 5 finished after 161.82µs Well a few weird things happen here. First, rounds 2, 3, and 4 don't print anything. Seriously? It turns out that the random changes we made didn't affect any of the degree counts, we moved edges between nodes, preserving degrees. It can happen. The second weird thing is that in round 5, with only two edge changes we have six changes in the output! It turns out we can have up to eight. The degree eight gets turned back into a seven, and a five gets turned into a six. But: going from five to six changes the count for each, and each change requires two record differences. Eight and seven wer