Rama

https://redplanetlabs.com/docs/~/
Admin
Rama is a new programming platform that combines database and stream processing capabilities with a...
Tokens:136,129
Snippets:1,379
Trust Score:7.6
Update:1 month ago
Show doc for...
Tokens:
Raw
### Depot Subscription Starting from Beginning

Source: https://redplanetlabs.com/docs/~/clj-defining-modules.html

Configure a depot subscription to start processing from the beginning of the depot partition using the `:start-from` option.

```clojure
(source> *depot {:start-from :beginning} :> *data)
```

--------------------------------

### KafkaExternalDepot Example

Source: https://redplanetlabs.com/docs/~/integrating.html

An example demonstrating how to declare and use a Kafka external depot within a Rama module. This example utilizes the `rama-kafka` library to implement the `ExternalDepot` interface for Apache Kafka.

```APIDOC
## KafkaExternalDepot Example

This example demonstrates how to declare and use a Kafka external depot within a Rama module. It uses the `rama-kafka` library to implement the `ExternalDepot` interface for Apache Kafka.

### Module Definition

```java
public class KafkaExternalDepotExampleModule implements RamaModule {
  @Override
  public void define(Setup setup, Topologies topologies) {
    Map<String, Object> kafkaConfig = new HashMap<>();
    kafkaConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1.mycompany.com:9092,kafka2.mycompany.com:9092");
    setup.declareObject("*kafka", new KafkaExternalDepot(kafkaConfig, "myTopic"));

    StreamTopology s = topologies.stream("s");
    s.source("*kafka").out("*record")
     .each(Ops.PRINTLN, "Kafka record:", "*record");
  }
}
```

### Usage Notes

- External depots are declared using `setup.declareObject` similar to other task globals.
- They can be used as sources for topologies by referencing their declared variable name in a `.source` declaration.
- Multiple topologies can consume from the same external depot.
- The concepts of "partition indexes" and "offsets" are fundamental to external depots, analogous to how data is distributed and identified in regular depots.
```

--------------------------------

### KafkaExternalDepot Example Module

Source: https://redplanetlabs.com/docs/~/integrating.html

Example of setting up and using KafkaExternalDepot as a source for a Rama stream topology. Requires the rama-kafka library.

```java
public class KafkaExternalDepotExampleModule implements RamaModule {
  @Override
  public void define(Setup setup, Topologies topologies) {
    Map<String, Object> kafkaConfig = new HashMap<>();
    kafkaConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka1.mycompany.com:9092,kafka2.mycompany.com:9092");
    setup.declareObject("*kafka", new KafkaExternalDepot(kafkaConfig, "myTopic"));

    StreamTopology s = topologies.stream("s");
    s.source("*kafka").out("*record")
     .each(Ops.PRINTLN, "Kafka record:", "*record");
  }
}
```

--------------------------------

### Example of Mirror PState and Depot Interaction

Source: https://redplanetlabs.com/docs/~/partitioners.html

This example demonstrates launching two modules where the second module queries a PState from the first. It highlights how `hashPartition` directs computation to the correct task and sets the partition index for querying mirrored objects, especially when module task counts differ.

```java
public class MirrorPStateExample {
  public static class Module1 implements RamaModule {
    @Override
    public void define(Setup setup, Topologies topologies) {
      setup.declareDepot("*depot", Depot.hashBy(Ops.IDENTITY));

      StreamTopology s = topologies.stream("s");
      s.pstate("$$p", PState.mapSchema(String.class, Long.class));

      s.source("*depot").out("*k")
       .compoundAgg("$$p", CompoundAgg.map("*k", Agg.count()));
    }
  }

  public static class Module2 implements RamaModule {
    @Override
    public void define(Setup setup, Topologies topologies) {
      setup.declareDepot("*depot", Depot.random());
      setup.clusterPState("$$other", Module1.class.getName(), "$$p");

      StreamTopology s = topologies.stream("s");
      s.pstate("$$p", PState.mapSchema(String.class, Long.class));

      s.source("*depot").out("*tuple")
       .each(Ops.EXPAND, "*tuple").out("*k", "*v")
       .hashPartition("$$other", "*k")
       .localSelect("$$other", Path.key("*k").nullToVal(0L)).out("*n")
       .each(Ops.PLUS_LONG, "*v", "*n").out("*newv")
       .hashPartition("*k")
       .localTransform("$$p", Path.key("*k").termVal("*newv"));
    }
  }

  public static void main(String[] args) throws Exception {
    try(InProcessCluster cluster = InProcessCluster.create()) {
      RamaModule module1 = new Module1();
      cluster.launchModule(module1, new LaunchConfig(8, 4));
      RamaModule module2 = new Module2();
      cluster.launchModule(module2, new LaunchConfig(4, 4));
      String module1Name = module1.getClass().getName();
      String module2Name = module2.getClass().getName();

      Depot depot1 = cluster.clusterDepot(module1Name, "*depot");
      Depot depot2 = cluster.clusterDepot(module2Name, "*depot");
      PState pstate = cluster.clusterPState(module2Name, "$$p");

      depot1.append("a");
      depot1.append("a");

      depot2.append(Arrays.asList("a", 10));
      System.out.println("Val: " + pstate.selectOne(Path.key("a")));
    }
  }
}
```

--------------------------------

### Configure Microbatch Start Options

Source: https://redplanetlabs.com/docs/~/microbatch.html

Set the starting point for microbatch processing of depots. These options are applied only on the first encounter of a depot.

```java
mb.source("*depot", MicrobatchSourceOptions.startFromBeginning())
```

```java
mb.source("*depot2", MicrobatchSourceOptions.startFromOffsetAfterTimestamp(107740800000))
```

```java
mb.source("*depot3", MicrobatchSourceOptions.startFromOffsetAgo(10000, OffsetAgo.RECORDS))
```

```java
mb.source("*depot4", MicrobatchSourceOptions.startFromOffsetAgo(15, OffsetAgo.DAYS))
```

--------------------------------

### Example Query Results

Source: https://redplanetlabs.com/docs/~/paths.html

Displays the output of the two example queries, showing the filtered time buckets and revenue for Query 1, and the list of product keys for Query 2.

```text
Query 1: [[10 14] [11 12] [16 17]]
Query 2: ["all-about-eve" "inherit-the-wind" "paths-of-glory" "stangers-on-a-train" "the-best-years-of-our-lives"]
```

--------------------------------

### Multi-Subscriber PState Example in Java

Source: https://redplanetlabs.com/docs/~/pstates.html

This example demonstrates setting up a PState with multiple subscribers using ProxyState. It shows how changes to the PState trigger callbacks with different diff types. Ensure you have the necessary Rama dependencies to run this code.

```java
public class MultiSubscriberModule implements RamaModule {
  @Override
  public void define(Setup setup, Topologies topologies) {
    setup.declareDepot("*depot", Depot.hashBy(Ops.IDENTITY));

    StreamTopology s = topologies.stream("s");
    s.pstate("$$p", Map.class).global();
    s.source("*depot").out("*k")
     .localTransform("$$p", Path.key("*k").nullToVal(0).term(Ops.INC));
  }

  public static void main(String[] args) throws Exception {
    try(InProcessCluster cluster = InProcessCluster.create()) {
      RamaModule module = new MultiSubscriberModule();
      cluster.launchModule(module, new LaunchConfig(1, 1));
      String moduleName = module.getClass().getName();

      Depot depot = cluster.clusterDepot(moduleName, "*depot");
      PState p = cluster.clusterPState(moduleName, "$$p");

      depot.append("a");
      depot.append("b");
      depot.append("c");

      ProxyState<Map> proxyTop = p.proxy(Path.stay(), new ProxyState.Callback<Map>() {
        @Override
        public void change(Map newVal, Diff diff, Map oldVal) {
          System.out.println("'top' callback: " + newVal + ", " + diff + ", " + oldVal);
        }
      });

      ProxyState<Integer> proxyA = p.proxy(Path.key("a"), new ProxyState.Callback<Integer>() {
        @Override
        public void change(Integer newVal, Diff diff, Integer oldVal) {
          System.out.println("'a' callback: " + newVal + ", " + diff + ", " + oldVal);
        }
      });

      ProxyState<Integer> proxyB = p.proxy(Path.key("b"), new ProxyState.Callback<Integer>() {
        @Override
        public void change(Integer newVal, Diff diff, Integer oldVal) {
          System.out.println("'b' callback: " + newVal + ", " + diff + ", " + oldVal);
        }
      });

      depot.append("a");
      depot.append("a");
      depot.append("c");
      depot.append("b");

      Thread.sleep(50);
    }
  }
}
```

--------------------------------

### Combine Start Options with Retry Policy

Source: https://redplanetlabs.com/docs/~/stream.html

Start options can be chained with retry policies, such as `retryNone()`, using the builder pattern. This allows for customized error handling alongside specific start conditions.

```java
s.source("*depot", StreamSourceOptions.startFromBeginning().retryNone())
```

--------------------------------

### limitAgg Output Example

Source: https://redplanetlabs.com/docs/~/aggregators.html

This shows the output of the limitAgg example, illustrating the processed data after applying aggregation and sorting options.

```text
Post agg data: 0 1000 1000
Post agg data: 1 10 12
Post agg data: 2 6 5
Query: [1016 1017]

```

--------------------------------

### Output of Materialize Example

Source: https://redplanetlabs.com/docs/~/intermediate-dataflow.html

The example demonstrates that the temporary PState materialized in the first batch block is correctly consumed and aggregated in the subsequent batch blocks, resulting in the sum of values.

```text
Vals: 6 9
```

--------------------------------

### Start Stream Processing from Beginning

Source: https://redplanetlabs.com/docs/~/stream.html

Use `startFromBeginning()` to process all records in a depot partition from the very first one. This option applies only on the initial run for a partition.

```java
s.source("*depot", StreamSourceOptions.startFromBeginning())
```

--------------------------------

### CaptureNewValInto Example with Multiple Keys/Aggregators

Source: https://redplanetlabs.com/docs/~/aggregators.html

Demonstrates the usage of `captureNewValInto` with multiple keys and/or multiple aggregators. This example is a placeholder for a more detailed implementation.

```java
public class CaptureNewValIntoModule implements RamaModule {
  @Override
  public void define(Setup setup, Topologies topologies) {

```

--------------------------------

### All Partitioning Example in Rama

Source: https://redplanetlabs.com/docs/~/partitioners.html

Illustrates using `allPartition` to send data to every task in the module. This is useful for actions that need to occur on all tasks, but use with caution as it can be resource-intensive.

```java
public class AllPartitionerExampleModule implements RamaModule {
  @Override
  public void define(Setup setup, Topologies topologies) {
    setup.declareDepot("*depot", Depot.random());

    StreamTopology s = topologies.stream("s");
    s.source("*depot").out("*k")
     .each(Ops.PRINTLN, "Start task", "*k", new Expr(Ops.CURRENT_TASK_ID))
     .allPartition()
     .each(Ops.PRINTLN, "End task", "*k", new Expr(Ops.CURRENT_TASK_ID));
  }

  public static void main(String[] args) throws Exception {
    try(InProcessCluster cluster = InProcessCluster.create()) {
      RamaModule module = new AllPartitionerExampleModule();
      cluster.launchModule(module, new LaunchConfig(4, 4));
      String moduleName = module.getClass().getName();

      Depot depot = cluster.clusterDepot(moduleName, "*depot");

      depot.append("groucho");
      depot.append("harpo");
      depot.append("chico");
    }
  }
}
```

--------------------------------

### Depot Subscription Starting After Timestamp

Source: https://redplanetlabs.com/docs/~/clj-defining-modules.html

Configure a depot subscription to start processing from the first record after a given timestamp using `offset-after-timestamp-millis` with `:start-from`.

```clojure
(source> *depot {:start-from (offset-after-timestamp-millis 12345678)} :> *data)
```

--------------------------------

### Java Aggregation Example

Source: https://redplanetlabs.com/docs/~/intermediate-dataflow.html

Demonstrates appending elements to a depot and waiting for microbatch processing. This is typically used to set up data for aggregation.

```java
Depot depot = cluster.clusterDepot(moduleName, "*depot");

depot.append("a");
depot.append("b");
depot.append("a");
depot.append("g");
depot.append("d");
depot.append("d");

cluster.waitForMicrobatchProcessedCount(moduleName, "mb", 6);
    }
  }
}
```

--------------------------------

### Migration Output Example

Source: https://redplanetlabs.com/docs/~/pstates.html

Illustrates the expected output of the IPC test, showing PState values before and after a module update with a migration applied.

```text
a before update: 1
b before update: 2
c before update: 3
a after update: 1!
b after update: 2!
c after update: 3!
```

--------------------------------

### Depot Subscription Starting from Offset Ago

Source: https://redplanetlabs.com/docs/~/clj-defining-modules.html

Configure a depot subscription to start processing from a specified number of records or time in the past using `offset-ago` with `:start-from`.

```clojure
(source> *depot {:start-from (offset-ago 10 :records)} :> *data)
```

```clojure
(source> *depot {:start-from (offset-ago 14 :days)} :> *data)
```

```clojure
(source> *depot {:start-from (offset-ago 6 :months)} :> *data)
```

--------------------------------

### Rama Branching and Unifying Dataflow Example

Source: https://redplanetlabs.com/docs/~/tutorial4.html

This example demonstrates constructing a branching dataflow graph in Rama. It defines an operation to emit different order types and then chains operations to process and unify salad and pizza orders before printing the final order size.

```java
public static class Brancher implements RamaOperation0 {
        @Override
        public void invoke(OutputCollector collector) {
            collector.emitStream("pizzaOrders", 1);
            collector.emitStream("saladOrders", 10);
            collector.emitStream("saladOrders", 11);
        }
    }

    public static void unifyExample() {
        Block.each(new Brancher()).outStream("pizzaOrders", "pizzaAnchor1", "*pizzaOrderSize")
                                  .outStream("saladOrders", "saladAnchor1", "*saladOrderSize")
             .each(Ops.DEC, "*saladOrderSize").out("*orderSize")
             .anchor("saladAnchor2")
             .hook("pizzaAnchor1")
             .each(Ops.INC, "*pizzaOrderSize").out("*orderSize")
             .anchor("pizzaAnchor2")
             .unify("pizzaAnchor2", "saladAnchor2")
             .each(Ops.PRINTLN, "*orderSize")
             .execute();
    }
```

--------------------------------

### Unification Output Example

Source: https://redplanetlabs.com/docs/~/clj-dataflow-lang.html

Prints the unified values from multiple dataflow branches.

```text
Val: 1
Val: 2
```

--------------------------------

### Java Shadowing Example

Source: https://redplanetlabs.com/docs/~/tutorial4.html

Demonstrates how a new binding with the same name does not affect other scopes.

```java
public static void shadowExample() {
    Block.each(Ops.IDENTITY, new Person("Megan")).out("*person")
         .each(new MultiOut()).outStream("stream1", "stream1Anchor")
                              .outStream("stream2", "stream2Anchor")
         .each(Ops.IDENTITY, new Person("John")).out("*person")
         .hook("stream1Anchor")
         .each((Person person) -> person.name).out("*personName")
         .each(Ops.PRINTLN, "*personName")
         .execute();
}
```

--------------------------------

### rama deploy with dynamic options

Source: https://redplanetlabs.com/docs/~/operating-rama.html

Example rama deploy command that sets dynamic options for each level using --moduleOptions, --topologyOptions, --depotOptions, and --pstateOptions. The syntax for module options is {dynamic option}={value};... and for the others is {entity},{dynamic option}={value};...

```bash
rama deploy \
--action launch \
--jar target/my-application.jar \
--module com.mycompany.MyModule \
--tasks 128 \
--threads 32 \
--workers 8 \
--replicationFactor 2 \
--moduleOptions 'replication.streaming.timeout.millis=20000;pstate.batch.buffer.limit=10' \
--topologyOptions 'myTopology,topology.combiner.limit=99' \
--depotOptions '*depot,depot.max.fetch=251' \
--pstateOptions '$$mb,pstate.reactivity.queue.limit=499;$$stream,pstate.reactivity.queue.limit=89'
```

--------------------------------

### Using customNav with a List Navigator

Source: https://redplanetlabs.com/docs/~/paths.html

Demonstrates how to use the `customNav` method to apply a custom navigator to a list. This example navigates to the first two elements of a list.

```java
List data1 = new ArrayList();
data1.add(10);
data1.add(11);
data1.add(12);
List data2 = new ArrayList();
data2.add(1);
Block.select(data1, Path.customNav(new MyListNav())).out("*v")
     .each(Ops.PRINTLN, "Val 1:", "*v")
     .select(data2, Path.customNav(new MyListNav())).out("*v")
     .each(Ops.PRINTLN, "Val 2:", "*v")
     .execute();
```

--------------------------------

### Select and Filter Values Using Path

Source: https://redplanetlabs.com/docs/~/paths.html

This example demonstrates selecting values from a map and then filtering them using filterLessThan. Only values less than 7 are printed.

```java
Block.select(data, Path.key("a0").mapVals().all().filterLessThan(7)).out("*v")
     .each(Ops.PRINTLN, "Val:", "*v")
     .execute();
```

--------------------------------

### Colocated Query Topology Output

Source: https://redplanetlabs.com/docs/~/query.html

Example output from invoking colocated query topologies. Demonstrates the results of calling 'q2' with different inputs.

```text
Query 1: 4
Query 2: 31
```

--------------------------------

### Rama Query Topology Example

Source: https://redplanetlabs.com/docs/~/tutorial2.html

Illustrates a complex query topology for finding 'followers you know'. This topology uses operations like `.out` and `.hashPartition` similar to ETLs.

```java
topologies.query("followersYouKnow", "*you", "*otherUser").out("*userSet")
          .hashPartition("$$following", "*you")
          .localSelect("$$followingById", Path.key("*you").sortedMapRangeFrom(0, 300).mapVals()).out("*following")
          .hashPartition("$$followers", "*otherUser")
          .localSelect("$$followers", Path.must("*otherUser", "*following"))
          .originPartition()
          .agg(Agg.set("*following")).out("*userSet");
```

--------------------------------

### Conductor rama.yaml configuration

Source: https://redplanetlabs.com/docs/~/operating-rama.html

Example Conductor configuration in rama.yaml. Sets the required zookeeper.servers list and the recommended local.dir for storing local state. Other settings are optional with defaults.

```yaml
zookeeper.servers:
  - "1.2.3.4"
  - "5.6.7.8"
local.dir: "/data/rama"
```

--------------------------------

### Supervisor configuration in rama.yaml

Source: https://redplanetlabs.com/docs/~/operating-rama.html

Example Supervisor configuration for rama.yaml. Required fields are conductor.host and zookeeper.servers; local.dir is optional but recommended. supervisor.port.range defaults to 3000, 4000 and should be a large range.

```yaml
zookeeper.servers:
  - "1.2.3.4"
  - "5.6.7.8"
conductor.host: "9.10.11.12"
local.dir: "/data/rama"
supervisor.port.range:
  - 2500
  - 3500
```

--------------------------------

### Retrieve a Page of Friends

Source: https://redplanetlabs.com/docs/~/tutorial6.html

Fetches a paginated list of friends for a given user, returning up to 20 friends at a time. Use an empty string for `start` to get the first page. The `excludeStart` option prevents duplicate entries when paginating.

```java
public Set<String> getFriendsPage(String userId, String start) {
  return _friends.selectOne(Path.key(userId).sortedSetRangeFrom(start, SortedRangeFromOptions.maxAmt(20).excludeStart()));
}
```

--------------------------------

### Maven pom.xml for building an uberjar with rama-helpers

Source: https://redplanetlabs.com/docs/~/operating-rama.html

Example pom.xml using maven-assembly-plugin to build an uberjar. The rama dependency is set to 'provided' scope to exclude Rama from the jar, preventing a size increase of over 100x. The resulting jar is located at target/myapplication-jar-with-dependencies.jar after running 'mvn package'.

```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.mycompany</groupId>
    <artifactId>myapplication</artifactId>
    <version>1.0.0</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <repositories>
      <repository>
        <id>nexus-releases</id>
        <url>https://nexus.redplanetlabs.com/repository/maven-public-releases</url>
      </repository>
    </repositories>

    <dependencies>
        <dependency>
            <groupId>com.rpl</groupId>
            <artifactId>rama</artifactId>
            <version>1.9.0</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>com.rpl</groupId>
            <artifactId>rama-helpers</artifactId>
            <version>0.10.0</version>
        </dependency>
    </dependencies>

    <build>
        <finalName>myapplication</finalName>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.1.1</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>
```

--------------------------------

### Rama GroupBy Operator Example

Source: https://redplanetlabs.com/docs/~/aggregators.html

Demonstrates the 'groupBy' operator in Rama for aggregating tuples. It groups by a key, performs count and sum aggregations, and then finds the top two tuples by sum. Requires Rama module setup and a query client.

```java
public class GroupByModule implements RamaModule {
  private SubBatch aggregatedTuples(String tuplesVar) {
    Block b = Block.each(Ops.EXPLODE, tuplesVar).out("*tuple")
                   .each(Ops.EXPAND, "*tuple").out("*k", "*val")
                   .groupBy("*k",
                     Block.agg(Agg.count()).out("*count")
                          .agg(Agg.sum("*val")).out("*sum"));
    return new SubBatch(b, "*k", "*count", "*sum");
  }

  @Override
  public void define(Setup setup, Topologies topologies) {
    setup.declareDepot("*depot", Depot.random());

    topologies.query("q", "*tuples").out("*topTwoTuples")
              .subBatch(aggregatedTuples("*tuples")).out("*k", "*count", "*sum")
              .each(Ops.TUPLE, "*k", "*count", "*sum").out("*tuple")
              .originPartition()
              .agg(Agg.topMonotonic(2, "*tuple")
                      .idFunction(Ops.FIRST)
                      .sortValFunction(Ops.LAST)).out("*topTwoTuples");
  }

  public static void main(String[] args) throws Exception {
    try(InProcessCluster cluster = InProcessCluster.create()) {
      RamaModule module = new GroupByModule();
      cluster.launchModule(module, new LaunchConfig(2, 2));
      String moduleName = module.getClass().getName();

      QueryTopologyClient<Long> query = cluster.clusterQuery(moduleName, "q");

      System.out.println(
        "Query: " +
        query.invoke(
          Arrays.asList(
            Arrays.asList("apple", 1),
            Arrays.asList("banana", 9),
            Arrays.asList("apple", 6),
            Arrays.asList("plum", 1),
            Arrays.asList("plum", 1),
            Arrays.asList("plum", 1))));
    }
  }
}
```

```text
Query: [["banana" 1 9] ["apple" 2 7]]
```

--------------------------------

### Isolation scheduler config with monitoring module

Source: https://redplanetlabs.com/docs/~/operating-rama.html

Example isolation scheduler configuration in rama.yaml for the Conductor. References the monitoring module as 'monitoring' and assigns it 2 workers, alongside user modules.

```yaml
conductor.assignment.mode:
  type: isolation
  modules:
    monitoring: 2
    com.mycompany.MyModule: 8
    com.mycompany.AnotherModule: 26
```

--------------------------------

### Worker configuration overrides YAML

Source: https://redplanetlabs.com/docs/~/operating-rama.html

Example YAML file for the --configOverrides flag. Sets worker.child.opts to "-Xmx8192m" and lists two custom serializations.

```yaml
worker.child.opts: "-Xmx8192m"
custom.serializations:
  - "com.rpl.myapp.serialization.MySerialization"
  - "com.rpl.myapp.serialization.MyOtherSerialization"
```

--------------------------------

### Example of Using Mirrored Components Across Modules

Source: https://redplanetlabs.com/docs/~/module-dependencies.html

Demonstrates how Module2 utilizes mirrored PState, Depot, and Query Topologies declared in Module1. `StreamSourceOptions.startFromBeginning()` is crucial for depots to ensure data is processed after module synchronization. A `Thread.sleep` is included to allow Module2 to finish processing before the cluster shuts down.

```java
public class UsingMirrorsExample {
  public static class Module1 implements RamaModule {
    @Override
    public void define(Setup setup, Topologies topologies) {
      setup.declareDepot("*depot1", Depot.hashBy(Ops.IDENTITY));
      setup.declareDepot("*depot2", Depot.hashBy(Ops.IDENTITY));

      StreamTopology s = topologies.stream("s");
      s.pstate("$$p", PState.mapSchema(String.class, Long.class));
      s.source("*depot1").out("*k")
       .compoundAgg("$$p", CompoundAgg.map("*k", Agg.count()));

      topologies.query("qq", "*v1", "*v2").out("*res")
                .each(Ops.TIMES, new Expr(Ops.INC, "*v1"), "*v2").out("*res")
                .originPartition();
    }
  }

  public static class Module2 implements RamaModule {
    @Override
    public void define(Setup setup, Topologies topologies) {
      setup.clusterDepot("*depot2", Module1.class.getName(), "*depot2");
      setup.clusterPState("$$mirror", Module1.class.getName(), "$$p");
      setup.clusterQuery("*mirrorQuery", Module1.class.getName(), "qq");

      StreamTopology s = topologies.stream("s");
      s.source("*depot2", StreamSourceOptions.startFromBeginning()).out("*k")
       .select("$$mirror", Path.key("*k")).out("*count")
       .invokeQuery("*mirrorQuery", 3, 7).out("*queryResult")
       .each(Ops.PRINTLN, "Results:", "*count", "*queryResult");
    }
  }

  public static void main(String[] args) throws Exception {
    try(InProcessCluster cluster = InProcessCluster.create()) {
      RamaModule module1 = new Module1();
      cluster.launchModule(module1, new LaunchConfig(4, 4));
      String module1Name = module1.getClass().getName();
      cluster.launchModule(new Module2(), new LaunchConfig(2, 2));

      Depot depot1 = cluster.clusterDepot(module1Name, "*depot1");
      Depot depot2 = cluster.clusterDepot(module1Name, "*depot2");

      depot1.append("a");
      depot1.append("a");
      depot1.append("b");
      depot2.append("a");
      depot2.append("b");
      depot2.append("c");

      Thread.sleep(2000);
    }
  }
}
```

--------------------------------

### PState Query Example

Source: https://redplanetlabs.com/docs/~/rest.html

Example of querying PState information using the REST API with curl.

```APIDOC
## POST /rest/com.mycompany.MyModule/pstate/$$profiles/selectOne

### Description
Selects a single PState entry based on provided keys.

### Method
POST

### Endpoint
`/rest/com.mycompany.MyModule/pstate/$$profiles/selectOne`

### Request Body
- **keys** (array) - Required - An array of keys to select the PState entry.
  - Example: `["alice", "location"]`

### Response
#### Success Response (200)
- **result** (string) - The selected PState value.

#### Response Example
```json
"New York City, NY"
```
```

--------------------------------

### Basic PState Client Queries with select and selectOne

Source: https://redplanetlabs.com/docs/~/pstates.html

This example demonstrates launching a module in an InProcessCluster, adding test data to a PState, and performing queries using `select` and `selectOne`. It highlights the difference in return types between `select` (list of navigated values) and `selectOne` (sole navigated value).

```java
public class ClientBasicSelectExamplesModule implements RamaModule {
  @Override
  public void define(Setup setup, Topologies topologies) {
    setup.declareDepot("*depot", Depot.hashBy(Ops.FIRST));

    StreamTopology s = topologies.stream("s");
    s.pstate("$$p",
      PState.mapSchema(
        String.class,
        PState.setSchema(Long.class)));

    s.source("*depot").out("*tuple")
     .each(Ops.EXPAND, "*tuple").out("*k", "*v")
     .compoundAgg("$$p", CompoundAgg.map("*k", Agg.set("*v")));
  }

  public static void main(String[] args) throws Exception {
    try(InProcessCluster cluster = InProcessCluster.create()) {
      RamaModule module = new ClientBasicSelectExamplesModule();
      cluster.launchModule(module, new LaunchConfig(4, 4));
      String moduleName = module.getClass().getName();

      Depot depot = cluster.clusterDepot(moduleName, "*depot");
      PState p = cluster.clusterPState(moduleName, "$$p");

      depot.append(Arrays.asList("cagney", 1));
      depot.append(Arrays.asList("cagney", 7));
      depot.append(Arrays.asList("cagney", 3));
      depot.append(Arrays.asList("cagney", 8));
      depot.append(Arrays.asList("davis", 10));
      depot.append(Arrays.asList("davis", 12));
      depot.append(Arrays.asList("davis", 14));

      System.out.println(
        "select (davis): " +  p.select(Path.key("davis"))); 
      System.out.println("selectOne (davis): " + p.selectOne(Path.key("davis")));
      System.out.println(
        "select (cagney): " +
          p.select(Path.key("cagney").all().filterPred(Ops.IS_ODD)));
    }
  }
}
```

```text
select (davis): [#{12 14 10}]
selectOne (davis): #{12 14 10}
select (cagney): [7 1 3]
```

--------------------------------

### Depot Append Example

Source: https://redplanetlabs.com/docs/~/rest.html

Example of appending data to a depot using the REST API with curl.

```APIDOC
## POST /rest/com.mycompany.MyModule/depot/*registerDepot/append

### Description
Appends data to a specified depot.

### Method
POST

### Endpoint
`/rest/com.mycompany.MyModule/depot/*registerDepot/append`

### Request Body
- **data** (object) - Required - The data payload to append.
  - **userid** (string) - Required - The ID of the user.

### Request Example
```json
{
  "data": {
    "userid": "alice"
  }
}
```

### Response
#### Success Response (200)
- **profileTopology** (string) - Indicates the success of the operation.

#### Response Example
```json
{
  "profileTopology": "success"
}
```
```

--------------------------------

### Conductor Redirect Example

Source: https://redplanetlabs.com/docs/~/rest.html

Example of accessing the REST API via the Conductor, which may return a redirect.

```APIDOC
## POST /rest/com.mycompany.MyModule/pstate/$$profiles/selectOne (via Conductor)

### Description
Accessing the REST API through the Conductor. The Conductor may return a 308 redirect to the appropriate Supervisor.

### Method
POST

### Endpoint
`/rest/com.mycompany.MyModule/pstate/$$profiles/selectOne`

### Request Body
- **keys** (array) - Required - An array of keys to select the PState entry.
  - Example: `["alice", "location"]`

### Response
#### Success Response (200)
- **result** (string) - The selected PState value.

#### Response Example
```json
"New York City, NY"
```

### Notes
Use the `-L` flag with `curl` to follow redirects. For high-throughput requests, it is recommended to use the Supervisor locations provided in the `Supervisor-Locations` header from the redirect instead of repeatedly querying the Conductor.
```

--------------------------------

### Define and Run HelloWorldModule

Source: https://redplanetlabs.com/docs/~/tutorial1.html

Defines a Rama module that prints 'Hello, world!!' to the console. This example demonstrates setting up an in-process cluster, launching a module, and appending data to a depot.

```java
package rama.examples.tutorial;

import com.rpl.rama.*;
import com.rpl.rama.module.*;
import com.rpl.rama.ops.Ops;
import com.rpl.rama.test.*;

public class HelloWorldModule implements RamaModule {
  @Override
  public void define(Setup setup, Topologies topologies) {
    setup.declareDepot("*depot", Depot.random());
    StreamTopology s = topologies.stream("s");
    s.source("*depot").out("*data")
     .each(Ops.PRINTLN, "*data");
  }

  public static void main(String[] args) throws Exception {
    try (InProcessCluster cluster = InProcessCluster.create()) {
      cluster.launchModule(new HelloWorldModule(), new LaunchConfig(1, 1));

      String moduleName = HelloWorldModule.class.getName();
      Depot depot = cluster.clusterDepot(moduleName, "*depot");
      depot.append("Hello, world!!");
    }
  }
}
```

--------------------------------

### Query Topology Invoke Example

Source: https://redplanetlabs.com/docs/~/rest.html

Example of invoking a query topology with arguments using the REST API with curl.

```APIDOC
## POST /rest/com.mycompany.MyModule/query/usersWithAge/invoke

### Description
Invokes a query topology with specified arguments.

### Method
POST

### Endpoint
`/rest/com.mycompany.MyModule/query/usersWithAge/invoke`

### Request Body
- **arguments** (array) - Required - An array of arguments for the query.
  - Example: `[31]`

### Response
#### Success Response (200)
- **result** (array) - A list of results from the query.
  - Example: `["cagney", "davis", "tracy", "lemmon"]`

#### Response Example
```json
["cagney", "davis", "tracy", "lemmon"]
```
```

--------------------------------

### Execute Reach Query in Main Method

Source: https://redplanetlabs.com/docs/~/query.html

Sets up an in-process cluster, launches the ReachModule, and populates depots with sample data. It then invokes the 'reach' query for 'grapefruit.com' and prints the result.

```java
public static void main(String[] args) throws Exception {
    try(InProcessCluster cluster = InProcessCluster.create()) {
      RamaModule module = new ReachModule();
      cluster.launchModule(module, new LaunchConfig(2, 2));
      String moduleName = module.getClass().getName();

      Depot urlsDepot = cluster.clusterDepot(moduleName, "*urlsDepot");
      Depot followsDepot = cluster.clusterDepot(moduleName, "*followsDepot");

      QueryTopologyClient<Integer> reach = cluster.clusterQuery(moduleName, "reach");

      urlsDepot.append(Arrays.asList("grapefruit.com", "jamescagney"));
      urlsDepot.append(Arrays.asList("grapefruit.com", "maeclarke"));
      followsDepot.append(Arrays.asList("jamescagney", "joanleslie"));
      followsDepot.append(Arrays.asList("jamescagney", "henryfonda"));
      followsDepot.append(Arrays.asList("jamescagney", "arlenefrancis"));
      followsDepot.append(Arrays.asList("jamescagney", "jacklemmon"));
      followsDepot.append(Arrays.asList("maeclarke", "henryfonda"));
      followsDepot.append(Arrays.asList("maeclarke", "charleslaughton"));
      followsDepot.append(Arrays.asList("maeclarke", "joanleslie"));
      followsDepot.append(Arrays.asList("maeclarke", "debbiereynolds"));

      cluster.waitForMicrobatchProcessedCount(moduleName, "core", 10);

      System.out.println("grapefruit.com reach: " + reach.invoke("grapefruit.com"));
    }
  }
}
```

--------------------------------

### Unordered Branching Output Example 1

Source: https://redplanetlabs.com/docs/~/clj-dataflow-lang.html

This is one of two possible outputs for the branching dataflow example, demonstrating nondeterministic ordering.

```text
Default output stream 1: 6
Default output stream 2: 6
Default output stream 1: 50
Default output stream 2: 50
Other output stream: 4
```

--------------------------------

### Equivalent of select> with hash partitioner

Source: https://redplanetlabs.com/docs/~/clj-dataflow-lang.html

Demonstrates the equivalent operations when `$$p` is configured with the default hash-based key partitioner.

```clojure
(|hash$$ $$p *user-id)
(local-select> [(keypath *user-id) ALL] $$p :> *v)
```

--------------------------------

### Unordered Branching Output Example 2

Source: https://redplanetlabs.com/docs/~/clj-dataflow-lang.html

This is the second possible output for the branching dataflow example, illustrating the nondeterministic nature of unordered branches.

```text
Default output stream 2: 6
Default output stream 1: 6
Default output stream 2: 50
Default output stream 1: 50
Other output stream: 4
```

--------------------------------

### Creating Custom Partitioners

Source: https://redplanetlabs.com/docs/~/clj-dataflow-lang.html

This example shows how to define and use a custom partitioner using '|custom'. Custom partitioners take a function that determines the target partition based on provided arguments and the number of partitions.

```clojure
(<<ramafn %custom
  [*num-partitions *arg1 *arg2]
  (:> (mod (+ *arg1 *arg2) *num-partitions)))
(|custom %custom 10 -11)
```

--------------------------------

### Declaring PStates with Options

Source: https://redplanetlabs.com/docs/~/clj-defining-modules.html

Shows how to declare PStates with various options: `:global?` for a single partition, `:initial-value` for a default starting value, `:private?` for restricted access, and `:key-partitioner` for custom partitioning logic.

```clojure
(defn my-partitioner [num-partitions key]
  (cond
    (= :a key) 0
    (= :b key) 1
    :else (mod (hash key) num-partitions)))

(defmodule Foo [setup topologies]
  (let [s (stream-topology topologies "s")]
    (declare-pstate s $$p1 Long {:global? true
                                 :initial-value 100})
    (declare-pstate s $$p2 {String Long} {:private? true})
    (declare-pstate s $$p3 {String Long} {:key-partitioner my-partitioner})
    ))
```

--------------------------------

### Get sequence of results with local-select>

Source: https://redplanetlabs.com/docs/~/clj-dataflow-lang.html

Wraps the path in `subselect` to get a sequence of results from `local-select>`, as it normally emits per navigated value.

```clojure
(local-select> (subselect (keypath *user-id) ALL) $$p :> *v)
```

--------------------------------

### Fetch Foreign Clients

Source: https://redplanetlabs.com/docs/~/clj-defining-modules.html

Demonstrates fetching clients for depot, PState, and query topologies from a foreign module using an opened cluster manager.

```clojure
(def manager (open-cluster-manager {"conductor.host" "1.2.3.4"}))
(def depot (foreign-depot manager "com.mycompany/MyModule" "*depot"))
(def pstate (foreign-pstate manager "com.mycompany/MyModule" "$$p"))
(def query (foreign-query manager "com.mycompany/MyModule" "my-query"))

```

--------------------------------

### Output of Circular Dependencies Example

Source: https://redplanetlabs.com/docs/~/module-dependencies.html

This output shows the results of running the `main` method in the `CircularDependenciesExample` class, demonstrating that the mirrored counts are correctly aggregated and displayed after establishing circular dependencies.

```text
Mirror count: a 2
Mirror count: b 1
```

--------------------------------

### Start Stream Processing from Records Ago

Source: https://redplanetlabs.com/docs/~/stream.html

Use `startFromOffsetAgo(count, OffsetAgo.RECORDS)` to start processing from a specific number of records prior to the current end of the partition. This allows for reprocessing a recent batch of data.

```java
s.source("*depot3", StreamSourceOptions.startFromOffsetAgo(10000, OffsetAgo.RECORDS))
```

--------------------------------

### Define and Invoke a Simple Query Topology

Source: https://redplanetlabs.com/docs/~/query.html

Defines a query topology named 'q' that takes two integer inputs and returns their sum plus one. The main method demonstrates how to launch the module and invoke the query topology from a client.

```java
public class SimpleQueryModule implements RamaModule {
  @Override
  public void define(Setup setup, Topologies topologies) {
    topologies.query("q", "*a", "*b").out("*res")
              .each(Ops.PLUS, "*a", "*b", 1).out("*res")
              .originPartition();
  }

  public static void main(String[] args) throws Exception {
    try(InProcessCluster cluster = InProcessCluster.create()) {
      RamaModule module = new SimpleQueryModule();
      cluster.launchModule(module, new LaunchConfig(4, 4));
      String moduleName = module.getClass().getName();

      QueryTopologyClient<Integer> q = cluster.clusterQuery(moduleName, "q");

      System.out.println("Query 1: " + q.invoke(1, 2));
      System.out.println("Query 2: " + q.invoke(10, 7));
    }
  }
}
```

--------------------------------

### Range Query from Key with Max Elements

Source: https://redplanetlabs.com/docs/~/clj-defining-modules.html

Utilize `sorted-map-range-from` to select a submap starting from a given key, up to a maximum number of elements. The `:inclusive?` option determines if the starting key is included, and `:max-amt` sets the element limit.

```clojure
(foreign-select-one (sorted-map-range-from :k 10) pstate)
```

```clojure
(foreign-select-one (sorted-map-range-from :a {:inclusive? false
                                               :max-amt 100})
                    pstate)
```

--------------------------------

### Java PState Reactive Query with Callback

Source: https://redplanetlabs.com/docs/~/pstates.html

This example demonstrates setting up a reactive query using PState and Depot. It includes a callback that is invoked when the PState value changes, receiving the new value, diff, and old value. Callbacks are executed in a background thread pool, ensuring ordered processing of diffs.

```java
      Depot depot = cluster.clusterDepot(moduleName, "*depot");
      PState p = cluster.clusterPState(moduleName, "$$p");

      depot.append(Arrays.asList("a", "b"));
      depot.append(Arrays.asList("a", "c"));

      ProxyState<Map> proxy = p.proxy(Path.key("a"), new ProxyState.Callback<Map>() {
        @Override
        public void change(Map newVal, Diff diff, Map oldVal) {
          System.out.println("Received callback: " + newVal + ", " + diff + ", " + oldVal);
        }
      });
      depot.append(Arrays.asList("a", "d"));
      depot.append(Arrays.asList("a", "c"));

      Thread.sleep(50);
    }
  }
}
```

--------------------------------

### Manual Cluster Backup (Free License)

Source: https://redplanetlabs.com/docs/~/backups.html

Manual steps to back up a Rama cluster when using the free license. This process requires taking the cluster offline and involves manual state copying.

```bash
rama shutdownCluster
```

--------------------------------

### limitAgg Options Example

Source: https://redplanetlabs.com/docs/~/aggregators.html

This example demonstrates the usage of various limitAgg options including sort, reverse, and indexVar. The sort option orders data by specified variables before selection, with reverse controlling ascending/descending order. indexVar binds a variable for post-aggregation indexing.

```java
}
  }
}
```
```
Tokens:
Raw