Dataflow programming

Dataflow programming can be understood in contrast to imperative programming, the industry-dominant paradigm that you probably have experience with.

You can think of imperative programming as defining a sequence of instructions to a worker for them to carry out: add these numbers, assign the result to this variable, call this method. You can see this physically embodied in something like a robot, where commands direct the robot’s behavior: turn to the right, walk forward four feet, turn to the left.

By contrast, dataflow programming is like constructing an assembly line. Each worker on the line waits for an input, performs an operation on it, and sends zero or more outputs to zero or more other workers. You can represent the relationships between workers as a directed graph, and you can refer to this graph as a dataflow pipeline. In Rama, we call these graphs topologies.

Dataflow programming can feel weird at first. Our usual way of constructing work is to give a kind of ontological primacy to subjects over verbs. It’s natural to think of the world as consisting of agents performing actions rather than as processes that entities flow through. It somehow feels more natural to think of a robot carrying out commands then it does to think of a network of nodes modifying data as it "flows" through.

Weird or not, it’s a great model for distributed programming. Dataflow is an extremely expressive way to map code to multiple cores or machines, which is why the Rama topology API was created.

Once again, we can understand the dataflow paradigm by comparing it to the imperative paradigm.

When an imperative program runs, its main thread of execution will start by calling some method (or function if you’re not using an OO language; since Rama is built on Java, let’s stick to talking about methods). Let’s call this Method A. Method A might in turn call Method B, which calls Method C, and so forth. Eventually, each of the method calls in this nested chain will return a value, until Method A returns a value. The main thread of execution will then move forward and call the next method, Method A', and the same process will play out.

Our mental model for an imperative program consists of caller methods coordinating with callee methods by passing in arguments and waiting for a single return value. This relationship is so ingrained that we can take it for granted. In fact, it may feel strange to draw attention to it or to refer to the relationship as "coordination", but it’s helpful to look at it that way because it helps with understanding the dataflow paradigm.

On an abstract level, the process of programming involves decomposing a system into functional units like methods, and then getting those units to coordinate in some way. With imperative languages, that coordination happens via the callstack. Because it happens implicitly, and because it’s so pervasive, it’s easy to gain the sense that it’s "the" way that programs work.

However, it’s not the only way for you to coordinate functional units. In a dataflow program, you think in terms of operations that wait for input. Upon receiving input, they perform some work and emit output values to one or more other operations.

The corresponding program could be written so that when Operation A receives an input, it emits two output values to Operation B and one output value Operation C. Operation B would perform its work for each of the two inputs it receives, and Operation C would perform its work the single value it receives. These operations don’t return a value to Operation A.

There are already some pronounced differences with an imperative program:

So in dataflow programs, you construct a computation graph where each node receives input and emits output values to other nodes. How does this all work in Rama? How do you specify the structure of the dataflow graph, and how do you specify the operations for each node?

These two questions are intertwined, but let’s start by focusing on how to specify the graph structure. (As we specify the graph, we’ll also specify the nodes' operations by using Rama’s built-in operations, but soon you’ll see how to write operations of your own.)

The examples we’ll look at will be very simple. They’ll be disconnected from modules and ETLs so that you can focus just on the dataflow machinery. The following code can be run from any Java process just with Rama on the classpath as a library. You can also try these examples in a Groovy shell – Groovy is similar enough to Java that these examples can be run in Groovy without changes. As a reminder, to launch a Groovy shell from rama-examples run mvn compile gplus:shell.

Here’s the first example:

To run this in a Groovy shell, enter the following:

We create a dataflow graph consisting of one node. Then we execute it, printing the number 10.

How does this work? Block.each() establishes a new dataflow graph and creates a single node in it; in general, each() will append a node. You can think of it as meaning, "Each time this node receives input, execute the given operation with the given arguments." In this case, the given operation is Ops.PRINTLN, and the argument is 10.

The Rama API follows the builder pattern, so that when you call the .each() method it returns a value that you can use to chain methods and further build up the dataflow graph, or (as in the case of this example) call the execute() method.

You can think of execute() as "send an input to this dataflow graph". The input is empty, but nevertheless the behavior of the dataflow graph is "execute the given operation on every input". Therefore, the Ops.PRINTLN operation runs and prints 10.

In real production code, you won’t use the execute() method. We’re only using it here to synthetically send input events to a dataflow graph’s root node. In real code, you’ll source data from data depots, like we did in previous examples:

Here, the dataflow graph’s root is a node created by s.source("*wordDepot"). This node works by continually reading records from the "*wordDepot" depot and emitting them.

With that explained let’s return to constructing dataflow graphs:

It’s possible to create branching dataflow graphs, where a node A emits to nodes B and C, and B and C’s operations execute independently of each other. It’s also possible to unify nodes, so that two or more nodes emit to the same node.

Doing this in Rama’s dataflow API is a little interesting because you’re building a 2-dimensional structure (the graph) using a 1-dimensional tool (a sequence of dataflow API calls). Here’s a bare bones example:

This constructs a graph that looks like this:

You can add conditionals to your dataflow graph, so that when the conditional is true a provided sub-graph receives input. The following example shows how you might check whether the "*currentUser" matches the "*username":

This creates a graph that looks like this:

The method ifTrue here takes two arguments, a condition value and a block. It creates a new node in the dataflow graph with two output streams; the node emits to the false stream when the condition value is null or false, and true otherwise. The provided block receives input from the true stream.

Note that there’s an implicit unification: both branches of the condition end up emitting to the same node.

Usually you wouldn’t bind an explicit var for the condition value, but you would nest an expression like so:

This is equivalent to the previous example, and in fact behind the scenes compiles to the previous example. Expr can be used anywhere input arguments are taken and are useful for making the code more concise.

What if you want to include an else clause to a condition? To do that, you just pass in another block as the third argument to ifTrue:

This produces a graph like this:

Again, there’s an implicit unification here. That means it’s possible to introduce new var bindings, but for those bindings to be accessible post-unification, they must be present in each branch. So this code would fail to compile:

But this code compiles fine because "*a" is available on both branches of the conditional:

Rama checks the scope of all vars when compiling a Block and will never execute code that has an invalid var reference.

Now that you’ve looked at how to construct a dataflow graph, let’s have a look at how to write custom operations for nodes. There are two different kinds of implementations for operations: RamaFunction and RamaOperation. Let’s look at an example of a RamaFunction first:

Of note:

RamaFunction is the simpler, stripped-down implementation of operations. When you implement a RamaFunction, your invoke method should always return a value. And whereas RamaOperation implementations can emit multiple times and to multiple streams, a RamaFunction can only emit once. You should prefer a RamaFunction where possible because it’s more efficient. It’s also more concise in that you don’t need to use an OutputCollector.

A RamaOperation is more general than a RamaFunction, capable of emitting multiple times, doing work between emits, emitting to multiple streams, and emitting multiple fields in a single emit. Here’s an example demonstrating all of this:

This function could be invoked in dataflow code like this:

Here, MyOperation is implementing RamaOperation0, indicating it takes zero arguments. And yet, invoke takes an argument, OutputCollector collector. This is the object you use to emit arbitrary values to arbitrary streams, and you can see in the example we do indeed emit to three streams, taking in various amounts of values as input. When invoking a RamaOperation via the dataflow API, it’s important when using out or outStream you capture exactly the number of values emitted. Just like how you must call methods in Java with the correct number of input arguments, you must capture the outputs of Rama with the correct number of output bindings.

When a RamaOperation emits, the code connected to the corresponding outStream or out executed immediately. Control flow returns back into the RamaOperation only when that downstream code finishes executing.

Another way to write custom operations is to use lambdas. You could rewrite the ramaFunctionExample above like this:

Yet another approach is to use method references:

Method references are nice because they can be used from other Java code as well as directly in topologies. Lambdas and method references are less verbose than a RamaFunction and are generally preferred.

Because scope bindings can be emitted to multiple nodes, you might be wondering if it’s possible for one node to change another node’s bindings.

Good news: scope bindings are immutable. When a node emits a scope, nothing can change it between the time it’s emitted and the time it’s received as input by its child nodes. Even if you create a new binding with the same name, that creates a new immutable scope and doesn’t affect other scopes. This example demonstrates this idea:

The MultiOut operation emits to two separate streams. It also implicitly emits its scope, which includes a binding for "*person".

In shadowExample, the "stream2" branch creates a new Person and binds that object to the name "*person". This will never affect the binding in "stream1"; "stream1" will always print "Megan".

In shadowMutationExample, however, the "stream2" branch actually mutates the shared object that happens to be bound to *person in both branches. In this case, it’s possible for "stream1" to print "John", depending on the order in which the branches are executed. In this case, because MultiOut emits to "stream1" first, it will print "Megan". If MultiOut were changed to emit to "stream2" first, it would print "John".

Rama’s dataflow API also includes a looping construct. In fact, because Rama’s dataflow API has both condition and looping constructs, it is Turing-complete. Here’s an example of looping in Rama:

This prints out:

Here is the graph it produces:

This looks somewhat similar to the conditional graph you just saw. loopWithVars creates a node, which we’ll call the loop root. When it emits, its scope includes an initial binding for the var "*i". It emits to an ifTrue node which checks a base condition for whether to execute an iteration of the loop.

The loop’s control flow is implemented with continueLoop and emitLoop. continueLoop creates a new node that emits a value to the node loop root. The loop root uses this to kick off a new loop iteration, binding "*i" to the value it receives as input from continueLoop. Thus the initial execution of the loop has "*i" bound to the number 0, and the next execution has "*i" bound to 1 and so on until the value of "*i" is 5.

When emitLoop is called, flow control immediately goes to the out("*loopValue") node and what follows from that. When that branch of code finishes, flow control goes back to the next node after emitLoop. Think of what comes after the loop as a function that emitLoop calls (this is formally known as the loop’s "continuation"). If you have multiple emitLoop nodes in a loop, they will all emit to the same loopEmitter node, and whatever loopEmitter receives as input it then executes the following node.

The way to execute another iteration of a loop is to call continueLoop with the new values for the loop vars. This example executes the loop body five times. If continueLoop is not called, execution of the loop stops.

Note that these semantics are different than what you’re used to from looping constructs in languages like Java, where control flow moves past a loop once the loop finishes. In Rama, control flow moves past the loop only when emitLoop is called and this is before the loop has finished executing. A loop finishing in Rama does not move control flow past the loop.

There’s another dataflow method loop that works exactly like loopWithVars but doesn’t require a var declaration. When using this construct you would generally determine whether to continue looping based on some mutable state bound before the loop. Here’s an example:

This prints:

Note here there’s no out attached to the loop call – Rama inserts that implicitly when it’s not provided.

In this section, you learned the basics of dataflow programming. Rama’s dataflow API is expressive and flexible, giving you precise control over how data is processed and how logic branches and merges.

There are more dataflow features in the Rama API that weren’t covered in this section. These include batch computation (e.g. joins, aggregation) and facilities for separating dataflow code into reusable chunks. These features and more are covered in Intermediate dataflow programming.