Defining and using modules

The main way to define a module is with defmodule. defmodule defines a regular Clojure function with arguments setup and topologies, like so:

setup is used to declare depots and dependencies to depots, PStates, or query topologies in other modules. topologies is used to declare stream, microbatch, and query topologies.

A module is named according to the namespace in which it’s declared and the symbol used to define it. If MyModule above was defined in the namespace com.mycompany, the module name would be "com.mycompany/MyModule". You can override the module name like so:

The owning namespace is always part of the module name and can’t be overridden. In this case the module name would be "com.mycompany/OtherName".

In tests, a module’s name can be fetched with get-module-name.

You can also define an anonymous module with module. This can sometimes be useful when writing unit tests. Here’s an example:

An anonymous module will have an auto-generated name, but you can override it with the same options map.

Depots are distributed, durable, and replicated logs of data. They are sources for topologies and can be appended to either by clients outside a cluster or from within topologies.

Depots are declared with declare-depot. Here’s an example of a module with six depots:

Symbols beginning with * are variables in Rama dataflow code, and the declare-depot macro lets you declare a depot’s name as a symbol – just as it will be referred later in dataflow code.

The last argument to declare-depot is the depot partitioner. This determines on which partition data appended by a client goes. As described in this section, this enables you to maintain local ordering for entities.

The depot partitioners can be :random, hash-by, :disallow, or a custom partitioner. :random causes appended data to go to a random partition. hash-by chooses the target task based on the hash of the value extracted with the provided function. :disallow disables appends from clients and only allows data to be appended from within topologies. Custom partitioners are defined with defdepotpartitioner as an arbitrary Clojure function.

Note that the function reference in hash-by must specify either a keyword or a top-level var defined with defn.

The only option available to declare-depot is :global?, which specifies the depot should have only a single partition. When this option is set, the partitioning scheme is irrelevant.

Whereas depot allow a topology to process new data as it arrives, tick depots allow a topology to process based on the passage of time. A tick depot emits an event according to the desired frequency (in milliseconds). For example, here’s a tick depot that emits an event once a minute:

See this section for differences in how tick depots work in streaming versus microbatching.

Mirrors are references within a module to depots, PStates, or query topologies in other modules. See the page on Module dependencies for details on using mirrors and tradeoffs to consider.

Mirrors are declared in the Clojure API with mirror-depot, mirror-pstate, and mirror-query. Here are some examples:

The mirror object is referenced in subsequent topology code with the variable specified in the second argument.

Both colocated and mirror query topologies can be invoked from a topology with invoke-query.

A task global allows global state to be available to all tasks of a module. Task globals live on each task and do not need to be transferred from task to task during processing. Task globals can be constant values, or each instance of a task global can specialize itself to each task. Task globals can be used for everything from distributing large constant data (like an ML model), in-memory caches specific to each partition, or integrating external services like databases, queues, or monitoring systems (where the task global opens up a client to the external service on each task).

A task global is declared with declare-object. For example:

If the provided object implements TaskGlobalObject, each instance will be specialized on each task. See the full documentation on task globals for more details.

ETL ("Extract-Transform-Load") topologies process data from depots and can maintain any number of PStates along the way. They can be used for purposes other than materializing PStates, such as interacting with third-party APIs or other external systems. There are two types of ETLs, streaming and microbatching, with different performance and fault-tolerance characteristics. See the Stream topologies and Microbatch topologies pages for more information on them.

Stream and microbatch topologies are declared with stream-topology and microbatch-topology. With a topology object you can then declare PStates and define dataflow code to handle processing from one or more depots.

Here’s an example of a stream topology:

declare-pstate is used to declare PStates for a topology, and <<sources defines dataflow code to subscribe and process data from depots. These will both be explained more in depth later on this page.

This module defines a topology that consumes a depot containing strings and materializes a PState $$counts containing the number of times each string appeared in the depot.

This code receives each piece of data from *depot and binds it to the variable *data. The :> keyword separates the input from the output of the operation. It then uses the |hash partitioner to switch to the task storing counts for that string. Finally, it uses +compound to update the PState. +compound does compound aggregation, which expresses the update in the shape of what’s being written. At the leaf, it uses the +count aggregator to specify the actual update. The package com.rpl.rama.aggs contains many built-in aggregators.

Here’s an example of a microbatch topology doing the same thing:

Whereas streaming processes data as it arrives, microbatching processes data across every partition of a depot in a batch computation. Each microbatch iteration processes the data that accumulated in the last iteration. source> in a stream topology emits data directly, while source> in a microbatch topology emits an object representing the batch of data for the iteration.

Variables beginning with % are anonymous operations. Calling %microbatch causes all data for the iteration to be emitted across all partitions individually. So if the microbatch contains 500 pieces of data on each depot partition, %microbatch emits and binds the *data variable 500 times across every partition.

%microbatch can be called multiple times in a microbatch iteration, or it can be called as part of a smaller subbatch. You’ll see examples in this section.

PStates are partitioned, durable, and replicated indexes that are declared with declare-pstate. A PState is an arbitrary combination of arbitrary size data structures. Each PState is declared with a schema that determines what it stores and how.

A feature called "subindexing" allows nested data structures to be of huge size, even more than the memory available. A schema defines the data structures used, as well as which ones are subindexed.

Here are some examples of schemas:

As you can see, you can have subindexed structures within subindexed structures, and the top-level schema doesn’t have to be a map. The Long schema specifies that each partition of the PState is a simple Long value. A PState like that is useful for ID generation, for example.

PState schemas can be specified as either data structure literals, or with explicit use of map-schema, set-schema, vector-schema, or fixed-keys-schema. The explicit forms allow subindexing options to be specified.

Subindex options can be specified with either the :subindex? or :subindex-options keywords. Only one of the two options may be specified. The :subindex-options variants allows subindexing to be specified with size tracking turned off, like so:

Size tracking is turned on by default and enables the size of subindexed structures to be fetched extremely quickly, even if the structure has huge numbers of elements. However, Rama’s internal maintenance of the size does require reads to be done when writing to a structure to see if a key is new or not. If you don’t need to ever query the size of a subindexed structure, you can turn it off to improve write performance.

PStates also accept four options when declaring them. Here are examples of using them:

:global? specifies the PState to have only a single partition.

:initial-value specifies the initial value of each PState partition. It should only be used with top-level value schemas (e.g. Long or java.util.Map).

:private? specifies the PState should only be readable from topology code within the module definition and not from external clients. Trying to fetch a client for a private PState will throw an exception.

Finally, :key-partitioner affects how foreign PState queries function. The default key partitioner is a hash partitioner that chooses the target partition to query based on the hash of the key. This is one of the most common ways to partition a PState. You can customize how foreign PState queries are routed with :key-partitioner. :key-partitioner should be a reference to a top-level function of two arguments: the number of partitions of the PState, and the key. See this section for more information on how key partitioners work.

The <<sources macro defines ETL logic using Rama’s dataflow API.

A <<sources block has one or more depot subscriptions specified using source>. The code following a source> call is the dataflow logic for processing data from that depot.

Each call to source> can specify subscription options. There are two options available: :start-from and :retry-mode.

:start-from determines where a topology starts processing on each partition the first time it ever runs. Note that :start-from options have no effect if the topology has run before, like on a module update. :start-from can be set to:

Examples:

:retry-mode only pertains to stream topologies since microbatching always has strong exactly-once semantics. :retry-mode can be set to :individual, :all-after, or :none, and it defaults to :individual if not set. For example:

The semantics of these retry modes is documented in this section.

All the code in a <<sources block utilizes Rama’s dataflow API. The dataflow API is different than regular Clojure programming. Whereas a Clojure function is based on "call and response" – you call a function and get a single result back – dataflow is "call and emit". That is, you call an operation and it emits values to downstream code. Operations can emit one time, many times, or even zero times. They can also emit multiple fields per emit or emit to independent output streams. Dataflow operations also don’t have to emit synchronously – they can emit asynchronously on a completely different partition on a different machine.

The dataflow API is documented fully on the next page.

A query topology is a predefined, on-demand, realtime, distributed computation that can operate over any or all of your PStates and any or all of the partitions of your PStates.

Here’s an example of declaring a module with a single query topology:

Like ETLs, query topologies are defined using topologies. The definition takes in a query name, "q1" in this example, a vector of parameters, and then dataflow code defining the query.

This example takes in one argument named *arg and emits a result variable *res. The dataflow code must then define *res, and this value will be sent back to the caller of the query topology.

A query topology must use the |origin partitioner, and it must be the final partitioner used. |origin indicates to partition back to the task where the query topology started.

If the query topology definition starts with an appropriate partitioner, the "leading partitioner" optimization will kick in, and query invokes will be routed directly to the correct task. You can learn more about this optimization and other aspects of query topologies on the page about query topologies.

Query topologies can be invoked from any topology, whether ETL or query, using invoke-query. invoke-query can invoke colocated or mirror query topologies, and query topologies can also invoke themselves recursively.

The next section will show how to invoke query topologies from clients outside a Rama cluster.

The term "foreign" refers to work in Rama initiated outside the cluster, including depot appends, PState queries, and query topology invokes. Clients are fetched with a "cluster manager", which on a real cluster is created with open-cluster-manager or open-cluster-manager-internal. When you use an in-process cluster in a test context via create-ipc (as is documented more on the Clojure API testing page), the cluster object is also a cluster manager.

Here’s an example of using open-cluster-manager and fetching depot, PState, and query topology clients:

open-cluster-manager can also be configured through a rama.yaml file on the classpath. open-cluster-manager-internal can be used if you configure separate external/internal hostnames for nodes on your cluster, like on AWS. See the Operating Rama page for more details.

foreign-depot, foreign-pstate, and foreign-query each take in the module name and the name of the desired object as arguments. Note that the names are strings, not symbols.

In this section, you learned how to define and use modules, depots, topologies, and PStates. On the next page, you’ll learn all the details of using Rama’s dataflow API.