Hands on Rama, day 2: Rewrite CAS, finish basic C(R)UD

The applications that benefit the most from Rama are those with complex domain models, large scale, or both.

Rama integrates and generalizes data ingestion, processing, indexing, and querying. The "integrates" aspect means it encompasses a suite of responsibilities that aren’t normally packaged in a single tool. The "generalizes" part means Rama is unrestricted in the kinds of applications it can build, including applications that require huge scale. This means Rama can build complex application backends on its own in a small amount of code that previously required dozens of tools and a lot of code. In many cases the code difference is over 100x.

Rama enables you to build data systems. They answer questions with the appropriate timeliness, accuracy, latency, and throughput, and have some scalability.

The key breakthrough of Rama is generalizing and integrating every aspect of data systems to such an extent that you can build entire data systems end-to-end with just Rama.

Rama was created in response to the proliferation of low-cost distributed computing resources. The Rama platform provides an execution environment [the cluster] and a compiler for that environment that enables you to write distributed applications without having to manage the underlying resources.

If a cluster is the execution environment, a module is the executable. You create modules to do work. We can understand modules in terms of their definitions and their runtime instantiation.

The idea here is there’s a relationship between an execution environment and languages that let you write applications for that environment. Execution environments provide resource abstractions: OS’s provide processes, files, file systems, sockets, a universal IO model, etc; Rama provides depots, partitioned states, distributed computations etc. Languages give you a way to manipulate those resources.

But the broader conception of performing work in Rama is that you’re coordinating distributed tasks across an arbitrary number of processes and machines to produce useful, distributed views of data.

PStates are a little similar to tables in an RDBMS in that both are named, discrete data containers within a larger data management system. They differ from tables in that they are arbitrarily compound data structures of arbitrary size, meaning that you can use them to store as much data as you need, organized however you need.

PState capabilities include "subindexing", which allows inner data structures in PStates to be of huge size, and "fine-grained reactive queries", which allows for continuous queries that are pushed "diffs" about changes to the result of the query.

At a high level, Rama is designed to run multiple copies of a module — called tasks — and distribute work among these copies. When you deploy a module, you specify how many tasks to run and how to distribute the tasks across resources: you could run multiple tasks within one JVM process, within multiple JVM processes on a single machine, or within JVM processes across multiple machines. Tasks run on threads within a single JVM, and a single thread can run multiple tasks. These threads are called "task threads".

1 worker is exactly for 1 module, and the module may have multiple workers (on different JVMs).

Rama has no guarantees about the global order in which depot records are processed, so this hypothetical module design [of a random depo partitioner + ETL which hashes on the data] introduced the possibility for race conditions - any 2 updates for X will be relocated to the same partition, but in a random order, since they arrive to different partitions and the time it takes to relocate over the network to the target one has a random aspect.

Why distributed systems? For performance, resiliency, perhaps [not in the case of Rama] separation of concerns.

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.

Depot append uses AckLevel.ACK (which is the default if not specified). With this ack level the depot append call will only complete when all streaming topologies co-located with the depot have finished processing the data.

Batch blocks: while normal dataflow is rather imperative, this is partially declarative, at a higher abstraction level & Rama decides a sequence of operations. It offers functionality including inner joins, outer joins, two-phase aggregation, and the ability to coordinate around batches of computation. Query topologies are implicitly batch blocks, and you can use batch blocks in microbatch topologies, though not in stream ones. A batch block runs in three phases: the "pre-agg phase", the "agg phase", and the "post-agg phase" - depending on where you put an .agg or .compoundAgg call(s). A final partitioner, if any, must be declared before the agg phase. Subbatches allow batch blocks to consume the results of other batch blocks ⇒ can do aggregates of aggregates.

Aggregators: An alternative to Paths for updating PStates, which enable huge increases in performance and expressivity in some cases. Two kinds of aggregators: more generic accumulators and more limited combiners.

Contrary to paths, aggregators know how to initialize values that don’t exist. Combiners allow parallelization ("two-phase aggregation", for combiners in batch blocks / query topos, especially impactful for global agg). Accumulators may take any number of args, Combiners always take the current + one new.

Other batch features: capture which PState entries changed.

A batch feature mostly useful in queries: use .agg without a PState argument and with .out to get the aggregate. In microbatch, aggregators can also .out("$$aTmpPState") (similar to .materialize used there w/o aggs).

All query topologies must contain .originPartition (= back to the query’s original task) as the final partitioner of the computation. The .out must be emitted exactly once.

Without subindexing, the entire data structure will be stored and retrieved as a single value. This will get expensive once there’s even just a few hundred elements in it. Subindexing enables inner data structures to efficiently contain huge numbers of elements, even more than could fit into memory.

Query topologies are implicitly batch blocks. So when programming them you’re always thinking in terms of pre-agg, agg, and post-agg phases. All the power of batch blocks, including joins, subbatches, and two-phase aggregation, is available.