User Guide

Information models are often superior to APIs, and almost always better than DSLs. The hyper-flexibility of a data structure literal allows Onyx workflows and catalogs to be constructed at a distance, meaning on another machine, in a different language, by another program, etc.

The information model for an Onyx workflow has the distinct advantage that it’s possible to compile other workflow representations (perhaps a datalog or SQL query) into the workflow that Onyx understands. The Information Model chapter describes a target for data structure compilation.

To the extent that Onyx places data at the highest importance, very few Onyx constructs actually need to be generated at the same time as program deployment or peer registration. Programs can create workflows, drop them to a database, and pull them out at a later time without any problems.

Macros are a tremendously powerful tool, but are often inappropriate for end-user consumption of an API. Onyx goes beyond Storm’s defbolt and defspout by making vanilla Clojure functions shine. These functions need no context to execute and do not require any dynamic bindings. They receive all information that they need via parameters, which are injected by Onyx’s task lifecycles.

To the same point above, we want plain Clojure functions to be the building blocks for application logic. Onyx’s functions can be tested directly without any special test runner.

In general, your design is in trouble when you’ve reached for with-redefs or something along those lines to mock functions. Onyx places a high importance around programming against interfaces, and even more-so around putting space in-between small components with channels. Onyx programs can be tested in development mode, and moved to production mode with only a small configuration file change. If you’d like to change your input or output plugins, all you need to do is re-associate the catalog entry with something like an in-memory plugin. No interface mocking code required.

It’s particularly telling that many compute frameworks don’t offer an easy way to parameterize workflows. Onyx puts space between the caller and the function definition. Parameterize tasks inside the catalog, and update the catalog entry at will. Additionally, Onyx allows peers to spin up their own parameters at boot-up time.

Onyx uses the notion of a sentinel value to transparently switch between streaming and batching modes. This makes it really easy to be able to reuse the same code for both batch and streaming computations.

Clojure functions again serve as a huge win. Dire is a library that supports aspects, meaning you can keep your application logic airtight away from logging, preconditions, and error handling.

Onyx AOT’s absolutely nothing on your behalf. When you’re ready to stand your jar up, simply uberjar and start executing on the target machine. Hadoop and Storm cause dependency hell (In Storm’s case, you’re restricted to a particular version of Clojure because you’re locked in by the Executor) by providing their own dependencies on top of yours. Onyx won’t mess with your dependencies.

You can, however, AOT Onyx yourself to speed up compilation times.

This section describes how log entries are applied to the peer’s local replica. A log entry is a persistent, sequential znode. Its content is a map with keys :fn and :args. :fn is mapped to a keyword that finds this log entry’s implementation. :args is mapped to another map with any data needed to apply the log entry to the replica.

Peers begin with the empty state value, and local state. Local state maintains a mapping of things like the inbox and outbox - things that are specific to this peer, and presumably can’t be serialized as EDN.

Each peer starts a thread that listens for additions to the log. When the log gets a new entry, the peer calls onyx.extensions/apply-log-entry. This is a function that takes a log entry and the replica, and returns a new replica with the log entry applied to it. This is a value-to-value transformation.

A single peer begins with the empty replica ({}) and progressively applies log entries to the replica, advancing its state from one immutable value to the next.

A peer reads the first log entry and applies the function to its local replica, moving the replica into a state "as of" entry 0

Because application of functions from the log against the replica are deterministic and free of side effects, peers do not need to coordinate about the speed that each plays the log. Peers read the log on completely independent timelines

Peers effect change in the world by reacting to log entries. When a log entry is applied, the peer calls onyx.extensions/replica-diff, passing it the old and new replicas. The peer produces a value summarizing what changed. This diff is used in subsequent sections to decide how to react and what side-effects to carry out.

Next, the peer calls onyx.extensions/reactions on the old/new replicas, the diff, and its local state. The peer can decide to submit new entries back to the log as a reaction to the log entry it just saw. It might react to "submit-job" with "volunteer-for-task", for instance.

After a peer reads a log entry and applies it to the log replica, it will (deterministically!) react by appending zero or more log entries to the tail of the log.

Finally, the peer can carry out side-effects by invoking onyx.extensions/fire-side-effects!. This function will do things like talking to ZooKeeper or writing to core.async channels. Isolating side effects means that a subset of the test suite can operate on pure functions alone. Each peer is tagged with a unique ID, and it looks for this ID in changes to its replica. The ID acts very much like the object orientated "this", in that it uses the ID to differentiate itself to conditionally perform side effects across an otherwise uniformly behaving distributed system.

Aside from the log structure and any strictly data/storage centric znodes, ZooKeeper maintains another directory for pulses. Each peer registers exactly one ephemeral node in the pulses directory. The name of this znode is a UUID.

Peers will fail, or be shut down purposefully. Onyx needs to: - detect the downed peer - inform all peers that this peer is no longer executing its task - inform all peers that this peer is no longer part of the cluster

The messaging layer of Onyx employees the same technique that Apache Storm uses to achieve fault tolerance. Any errors are our own.

One of the primary obstacles that this design imposes is the requirement of seemingly infinite storage. Log entries are only ever appended - never mutated. If left running long enough, ZooKeeper will run out of space. Similarly, if enough jobs are submitted and either completed or killed, the in memory replica that each peer houses will grow too large. Onyx requires a garbage collector to be periodically invoked.

When the garbage collector is invoked, two things will happen. The caller of gc will place an entry onto the log. As each peer processed this log entry, it carries out a deterministic, pure function to shrink the replica. The second thing will occur when each peer invokes the side effects for this log entry. The caller will have specified a unique ID such that it is the only one that is allowed to trim the log. The caller will take the current replica (log entry N to this log entry), and store it in an "origin" znode. Anytime that a peer boots up, it first reads out of the origin location. Finally, the caller deletes log entry N to this log entry minus 1. This has the dual effect of making new peers start up faster, as they have less of the log to play. They begin in a "hot" state.

The garbage collector can be invoked by the public API function onyx.api/gc. Upon returning, the log will be trimmed, and the in memory replicas will be compressed.

A peer can start by reading out of the origin, and continue directly to a particular log location.

prepare-join-cluster

notify-join-cluster

accept-join-cluster

add-virtual-peer

abort-join-cluster

group-leave-cluster

leave-cluster

seal-task

submit-job

kill-job

gc

signal-ready

set-replica!

assign-bookkeeper-log-id

The Core API is used to start/stop resources, jobs, and monitor job progress. It’s accessible through the onyx.api namespace.

The Peer Pipeline API allows you to interact with data storage mediums to read and write data for plugins.

A Function is a construct that takes a segment as a parameter and outputs a segment or a seq of segments. Functions are meant to literally transform a single unit of data in a functional manner. The following is an example of a function:

Note that you may only pass segments between functions - no other shape of data is allowed.

A function can be parameterized before a job is submitted to Onyx. The segment is always the last argument to the function. There are multiple ways to paramerize a function, and they can be used in combination.

The function is then invoked with (partial f "abc" "def"). The order is controlled by the vector of :onyx/params.

The function is then invoked with (partial f 42).

Using this approach "hard sets" the parameters list. Other parameters may already exist in onyx.core/params. If you want to retain those parameter, concat them together and return the new value on onyx.core/params.

The function is then invoked with (partial f 64).

This approach is useful for parameterizing a task regardless of which job it is in. If both onyx.peer/fn-params and :onyx/params are set for the same task, they are concatenated together, with fn-params coming first.

Grouping ensures that "like" values are always routed to the same virtual peer, presumably to compute an aggregate. Grouping is specified inside of a catalog entry. There are two ways to group: by key of segment, or by arbitrary function. Grouping by key is a convenience that will reach into each segment and pin all segments with the same key value in the segment together. Grouping functions receive a single segment as input. The output of a grouping function is the value to group on. Grouped functions must set keys :onyx/min-peers and :onyx/flux-policy. See below for a description of these.

To group by a key or a vector of keys in a segment, use :onyx/group-by-key in the catalog entry:

To group by an arbitrary function, use :onyx/group-by-fn in the catalog entry:

Functions that use the grouping feature are presumably stateful. For this reason, unless :continue is used, once a job begins, no matter how many peers are added to the cluster, no new peers will be allocated to grouping tasks. When more peers are added after the job begins, the hashing algorithm loses its consistency, and stateful operations won’t work correctly.

Given the fact the Onyx will not add more peers to regular grouping tasks after it begins, we introduce a new parameter - :onyx/min-peers. This should be set to an integer that indicates the minimum number of peers that will be allocated to this task before the job can begin. Onyx may schedule more than the minimum number that you set. You can create an upper bound by also using :onyx/max-peers.

One concern that immediately needs to be handled is addressing what happens if a peer on a grouping task leaves the cluster after the job has begun? Clearly, removing a peer from a grouping task also breaks the consistent hashing algorithm that supports statefulness. The policy that is enforced is configurable, and must be chosen by the developer. We offer two policies, outlined below.

Sometimes you might be able to perform a function more efficiently over a batch of segments rather than processing one segment at a time, such as writing segments to a database in a non-output task. You can receive the entire batch of segments in bulk as an argument to your task by setting :onyx/bulk? to true in your catalog entry for your function. Onyx will ignore the output of your function and pass the same segments that you received downstream. The utility of this feature is that you receive the entire batch in one shot. Onyx ignores your output because it would make it impossible to track which specific messages are children of particular upstream messages - breaking Onyx’s fault tolerance mechanism.

Functions with this key enabled may not be used with flow conditions. These segments are passed to all immediate downstream tasks.

An example catalog entry:

And an example catalog function to correspond to this entry:

The default value for this option is false.

Sometimes you’re going to want a node in your workflow with no outgoing connections that doesn’t perform I/O against a database. You can do this by setting :onyx/type to :output, :onyx/medium to :function, and :onyx/plugin to onyx.peer.function/function. Then you can specify an :onyx/fn pointing to a regular Clojure function. For example:

Workflows specify the structure of your computation as a directed, acyclic graph. A workflow describes all possible routes that a segment can take as it enters your workflow. On the other hand, we often have the need to specify how an individual segment moves throughout your workflow. Many times, a segment conditionally moves from one task to another. This is a concept that Onyx takes apart and turns into its own idea, independent of the rest of your computation. They’re called Flow Conditions. It should be mentioned straight away that Flow Conditions are entirely optional, and your program can ignore them entirely if you’d like. Omitting them leads to the default behavior, which sends a segment to all immediate downstream tasks.

The easiest way to learn how to use flow conditions is to see an example. Suppose we have the following workflow snippet:

This workflow takes some input in (presumably a stream of people), and directs segments to four possible tasks - :process-children, :process-adults, :process-female-athletes, and :process-everyone. Suppose we want to conditionally direct a segment to zero or more of these tasks, depending on some predicates. We use flow conditions to carry out this work. Flow conditions are their own data structure that are bundled along with the workflow and catalog to onyx.api/submit-job (with key :flow-conditions). Here’s an example of what a flow conditions data structure would look like for our proposed workflow:

The basic idea is that every entry in the Flow Conditions data structure denotes a relationship between a task and its downstream tasks. :flow/from indicates the task that the segment is leaving, and :flow/to indicates the tasks that the segment should be sent to if the predicate evaluates to true. The predicate is denoted by :flow/predicate, which is a keyword or sequence of keywords that are resolved to a function. Later in this section, we’ll cover how exactly the predicate function is constructed.

There is one flow conditions data structure per job - that is, there is one vector of maps. The order that you specify the flow conditions in matters. More on that later in this section.

A predicate function is a Clojure function that takes at least four parameters - a context map, the old segment, the new segment, and the collection of all new segments produced from the old segment. Predicates can take parameters at runtime. They will be appended to the end of the function invocation. See Predicate Parameters for further discussion of how to use runtime parameters.

Predicates for the above examples can be seen below:

Predicate functions can take parameters at runtime. In this first flow condition, we use the parameter :my/max-child-age and set its value to 17. We pass this value to the predicate by surrounding it with brackets, as in: [:my.ns/child? :my/max-child-age]. The parameters are appended to the end of the function call to the predicate. See Predicate Function Signatures in this section to see the arguments that are passed into the predicate regardless each invocation.

Sometimes, the decision of whether to allow a segment to pass through to the next task depends on some side effects that were a result of the original segment transformation. Onyx allows you to handle this case by adding extra keys to your segment that comes out of the transformation function. These extra keys are visible in your predicate function, and then stripped off before being sent to the next task. You can indicate these "extra keys" by the setting :onyx/exclude-keys to a vector of keys.

For example, if we had the following transformation function:

Our predicate for flow conditions might need to use the :side-effects-result to make a decision. We don’t want to actually send that information over out to the next task, though - so we :flow/exclude-keys on :side-effects-results to make it disappear after the predicate result has been realized.

One extraordinarily powerful feature of Flow Conditions is its composition characteristics. Predicates can be composed with logical and, or, and not. We use composition to check if the segment is both female and an athlete in [:and :my.ns/female? :my.ns/athlete?]. Logical function calls must be surrounded with brackets, and may be nested arbitrarily. Functions inside of logical operator calls may be parameterized, as in [:and :my.ns/female? [:my.ns/child? :my/max-child-age]]. Parameters may not specify logical functions.

Sometimes, you want a flow condition that emits a value to all tasks if the predicate is true. You can use short hand to emit to all downstream tasks:

Similarly, sometimes you want to emit to no downstream tasks:

If a flow condition specifies :all as its :flow/to, it must come before any other flow conditions. If a flow condition specifies :none as its :flow/to, it must come directly behind an :all condition, or first if there is no :all condition. This is because of the semantics of short circuiting. We’ll discuss what short circuiting means later in this section.

:flow/to set to :all or :none must always set :flow/short-circuit? to true.

:flow/from may be set to :all. This directs all immediate upstream links to pass segments to this task’s flow condition. :flow/from as :all does not impose order constraints as :flow/to set to :all does.

If multiple flow condition entries evaluate to a true predicate, their :flow/to values are unioned (duplicates aren’t acknowledged), as well as their :flow/exclude-keys. Sometimes you don’t want this behavior, and you want to specify exactly the downstream tasks to emit to - and not check any more flow condition entries. You can do this with :flow/short-circuit? set to true. Any entry that has :flow/short-circuit? set to true must come before any entries for an task that have it set to false or nil.

Flow Conditions give you leverage for handling exceptions without miring your code in try/catch logic. If an exception is thrown from an Onyx transformation function, you can capture it from within your flow conditions by setting :flow/thrown-exception? to true. It’s default value is false. If an exception is thrown, only flow conditions with :flow/thrown-exception? set to true will be evaluated. The value that is normally the segment which is sent to the predicate will be the exception object that was thrown. Exception flow conditions must have :flow/short-circuit? set to true. Note that exceptions don’t serialize. This feature is meant to be used in conjunction with Post-transformations and Actions for sending exception values to downstream tasks.

And the predicate might be:

This will only restrict the flow from :input-stream to :error-task when an exception is thrown - see the discussion of Short Circuiting above. When an exception is not thrown, the default behaviour will apply. For example, if there are later flow conditions, they will apply. If not will flow through to all tasks if there are no other flow conditions for that task.

Post-transformations are extension provided to handle segments that cause exceptions to arise. If a flow condition has :flow/thrown-exception? set to true, it can also set :flow/post-transform to a keyword. This keyword must have the value of a fully namespace qualified function on the classpath. This function will be invoked with three parameters: the event map, the segment that caused the exception, and the exception object. The result of this function, which must be a segment, will be passed to the downstream tasks. This allows you to come up with a reasonable value to pass downstream when you encounter an exception, since exceptions don’t serialize anyway. :flow/exclude-keys will be called on the resulting transformed segment.

Example:

And an example post-transform function might be:

After a set of flow conditions has been evaluated for a segment, you usually want to send the segment downstream to the next set of tasks. Other times, you want to retry to process the segment because something went wrong. Perhaps a database connection wasn’t available, or an email couldn’t be sent.

Onyx provides Flow Conditions :flow/action to accomplish this. By setting :flow/action to :retry, a segment will expire from the internal pool of pending messages and be automatically retried from its input task. If any of the :flow/action`s from the matching flow conditions are `:retry, the segment will be retried and will not be sent downstream. This parameter is optional, and it’s default value is nil. nil will cause the segment to be sent to all downstream tasks that were selected from evaluating the flow conditions. Any flow condition clauses with :flow/action set to :retry must also have :flow/short-circuit? set to true, and :flow/to set to :none.

Here’s a quick example:

The messaging layer takes care of the direct peer to peer transfer of segment batches, acks, segment completion and segment retries to the relevant virtual peers.

The Onyx messaging implementation is pluggable and alternative implementations can be selected via the :onyx.messaging/impl Peer Configuration option.

There are several interesting points to execute arbitrary code during a task in Onyx. Onyx lets you plug in and calls functions before a task, after a task, before a batch, and after a batch on every peer. Additionally, there is another lifecycle hook that allows you to delay starting a task in case you need to do some work like acquiring a lock under contention. A peer’s lifecycle is isolated to itself, and lifecycles never coordinate across peers. Usage of lifecycles are entirely optional. Lifecycle data is submitted as a data structure at job submission time.

Let’s work with an example to show how lifecycles work. Suppose you want to print out a message at all the possible lifecycle hooks. You’d start by defining 9 functions for the 9 hooks:

Notice that all lifecycle functions return maps except start-task?. This map is merged back into the event parameter that you received. start-task? is a boolean function that allows you to block and back off if you don’t want to start the task quite yet. This function will be called periodically as long as false is returned. If more than one start-task? is specified in your lifecycles, they must all return true for the task to begin. start-task? is invoked before before-task-start.

Next, define a map that wires all these functions together by mapping predefined keywords to the functions:

Each of these 9 keys maps to a function. All of these keys are optional, so you can mix and match depending on which functions you actually need to use.

Finally, create a lifecycle data structure by pointing :lifecycle/calls to the fully qualified namespaced keyword that represents the calls map that we just defined. Pass it to your onyx.api/submit-job call:

It is also possible to have a lifecycle apply to every task in a workflow by specifying :lifecycle/task :all. This is useful for instrumenting your tasks with metrics, error handling, or debugging information.

You can supply as many sets of lifecycles as you want. They are invoked in the order that they are supplied in the vector, giving you a predictable sequence of calls. Be sure that all the keyword symbols and functions are required onto the classpath for the peer that will be executing them.

Windowing splits up a possibly unbounded data set into finite, possibly overlapping portions. Windows allow us create aggregations over distinct portions of a stream, rather than stalling and waiting for the entire data data set to arrive. In Onyx, Windows strictly describe how data is accrued. When you want to do something with the windowed data, you use a Trigger. See the chapter on Triggers for more information. Onyx’s windowing mechanisms are strong enough to handle stream disorder. If your data arrives in an order that isn’t "logical" (for example, :event-time keys moving backwards in time), Onyx can sort out the appropriate buckets to put the data in.

The topic of windows has been widely explored in the literature. There are different types of windows. Currently, Onyx supports Fixed, Sliding, Global, and Session windows. We will now explain the supported window types.

Onyx allows you to specify range and slide values in different magnitudes of units, so long as the units can be coverted to the same unit in the end. For example, you can specify the range in minutes, and the slide in seconds. Any value that requires units takes a vector of two elements. The first element represents the value, and the second the unit. For example, window specifications denoting range and slide might look like:

See the information model for all supported units. You can use a singular form (e.g. :minute) instead of the plural (e.g. :minutes) where it makes sense for readability.

Onyx is also capable of sliding by :elements. This is often referred to as "slide-by-tuple" in research. Onyx doesn’t require a time-based range and slide value. Any totally ordered value will work equivalently.

Windows allow you accrete data over time. Sometimes, you want to store all the data. Othertimes you want to incrementally compact the data. Window specifications must provide a :window/aggregation key. We’ll go over an example of every type of aggregation that Onyx supports.

See the Information Model chapter for an exact specification of what values the Window maps need to supply. Here we will describe what each of the keys mean.

Windows capture data over time and place segments into discrete, possibly overlapping buckets. By itself, this is a relatively useless concept. In order to harness the information that has been captured and rolled up, we need to move it somewhere. Triggers let us interact with the state in each extent of a window.

Onyx ships a number of trigger implementations that can be used out of the box. Each trigger fires in response to a particular stimulous. All triggers implemented in Onyx core fire at task completion. We outline each here and show an example of each in action.

A refinement mode allows you to articulate what should happen to the state of a window extent after a trigger has been invoked.

Onyx offers you the ultimate flexibility on what to do with your state during a trigger invocation. Set :trigger/sync to a fully qualified, namespaced keyword pointing to a function on the classpath at runtime. This function takes 5 arguments: The event map, the window map that this trigger is defined on, the trigger map, a state-event map, and the window state as an immutable value. Its return value is ignored.

This function is invoked when the trigger fires, and is used to do any arbitrary action with the window contents, such as sync them to a database. It is called once per window instance. In other words, if a fixed window exists with 5 instances, the firing of a Timer trigger will call the sync function 5 times. You can use lifecycles to supply any stateful connections necessary to sync your data. Supplied values from lifecycles will be available through the first parameter - the event map.

See the Information Model chapter for an exact specification of what values the Trigger maps need to supply. Here we will describe what each of the keys mean.

Onyx provides the ability to perform updates to a state machine for segments which are calculated over windows. For example, a grouping task may accumulate incoming values for a number of keys over windows of 5 minutes. This feature is commonly used for aggregations, such as summing values, though it can be used to build more complex state machines.

As segments are processed, an internal state within the calculated window is updated. In this case we are trying to sum the ages of the incoming segments.

Window aggregations are defined by a map containing the following keys:

In the :window/aggregation map in the :sum-all-ages window referenced above.

Let’s try processing some example segments using this aggregation:

Results in the following events:

This state can be emitted via triggers or another mechanism. By describing changelog updates as a vector with a log command, such as :set-value aggregation function can emit multiple types of state transition if necessary.

To allow for full recovery after peer crashes, the window state must be replicated somewhere. As state updates occur, Onyx publishes the stream of changelog updates to a replicated log.

After the changelog entry is written to the replicated log, the segment is acked, ensuring that a segment is only cleared from the input source after the update to window states it caused has been fully written to the log. When a peer crash occurs, a new peer will be assigned to the task, and this peer will play back all of the changelog entries, and apply them to the state, starting with the initial state. As the changelog updates are read back in the same order that they were written, the full state will be recovered. Partial updates ensure that only minimal update data is written for each segment processed, while remaining correct on peer failure.

Exactly once aggregation updates are supported via Onyx’s filtering feature. When a task’s catalog has :onyx/uniqueness-key set, this key is looked up in the segment and used as an ID key to determine whether the segment has been seen before. If it has previously been processed, and state updates have been persisted, then the segment is not re-processed. This key is persisted to the state log transactionally with the window changelog updates, so that previously seen keys can be recovered in case of a peer failure.

See the section "Exactly Once Side-Effects" for discussion of why side-effects are impossible to achieve Exactly Once.

In order to reduce memory consumption, uniqueness-key values are persisted to a local database, currently implemented with RocksDB. This database uses a bloom filter, and a memory cache, allowing Onyx to avoid hitting disk for most filter key checks.

In order to prevent unbounded increase in the size of the filter’s disk consumption, uniqueness-key values are bucketed based on recency, and the oldest bucket is expired as the newest is filled.

Several configuration parameters are available for the rocksdb based local filter. The most relevant of these for general configuration is :onyx.rocksdb.filter/num-ids-per-bucket, and :onyx.rocksdb.num-buckets, which are the size and the number of buckets referenced above.

Exactly once side-effects resulting from a segment being processed may occur, as exactly once side-effects are impossible to achieve. Onyx guarantees that a window state updates resulting from a segment are perfomed exactly once, however any side-effects that occur as a result of the segment being processed cannot be guaranteed to only occur once.

State update changelog entries are persisted to BookKeeper, a replicated log server. An embedded BookKeeper server is included with Onyx. You can either use the embedded or run BookKeeper along side Onyx in a separate process.

BookKeeper ensures that changelog entries are replicated to multiple nodes, allowing for the recovery of windowing states upon the crash of a windowed task.

By default the the Onyx BookKeeper replication is striped to 3 BookKeeper instances (the quorum), and written to 3 instances (the ensemble).

The embedded BookKeeper server can be started via the onyx.api/start-env api call, with an env-config where :onyx.bookkeeper/server? is true.

When running on a single node, you may wish to use BookKeeper without starting the multiple instances of BookKeeper required to meet the ensemble and quorum requirements. In this case you may start a local quorum (3) of BookKeeper servers by setting :onyx.bookkeeper/local-quorum? to true.

Embedded BookKeeper Configuration Parameters

It is recommended that the state changelog is periodically compacted. When compaction occurs, the current state is written to a new ledger and all previous ledgers are swapped for the new compacted state ledger.

Compaction can currently only be performed within a task lifecycle for the windowed task. Be careful to choose the condition (see YOUR-CONDITION in the example below, as compacting too often is likely expensive. Compacting once every X segments is reasonable good choice of condition.

The BookKeeper state log implementation can be configured via the peer-config. Of particular note, is :onyx.bookkeeper/ledger-password which generally be changed to a more secure default.

Onyx eschews customized abstractions for describing distributed programs. As a consequence, application code written on top of Onyx ends up being regular Clojure functions and data structures. Onyx remains out of the way, and you’re free to test your functions as you would any other Clojure program. It’s also useful to roll up your entire job and test it as it will run in production. This section is dedicated to giving you a set of idioms and tools for working with Onyx jobs during the development phase.

Unlike Hadoop and Storm, Onyx does not have a built-in deployment feature. To deploy your application, you need to uberjar your program and place it on every node in your cluster. Start up the uberjar, passing it your shared Onyx ID and ZooKeeper address. Once it connects, you’ve successfully deployed!

We’ve chosen not to build in a deployment strategy because there are so many flexible approaches to handling deployment. S3, Docker, Mesos, Swarm, and Kubernetes are a few good choices. We’ll describe some of these strategies below.

When you’re taking Onyx to production, it’s not enough to know what your application-specific code is doing. You need to have insight into how Onyx is operating internally to be able to tune performance at an optimal level. Onyx allows you to register callbacks that are invoked when critical sections of code are executed in Onyx, returning a map that includes latency and data load size if appropriate.

Callback functions that are given to Onyx for monitoring take exactly two parameters. The first parameter is the monitoring configuration, the second is a map. The return value of the function is ignored. The event names and keys in their corresponding maps are listed in a later section of this chapter, as well as a discussion about monitoring configuration. Here’s an example of a callback function:

To register callbacks, create a map of event name key to callback function. This map must have a key :monitoring, mapped to keyword :custom. If you want to ignore all callbacks, you can instead supply :no-op. Monitoring is optional, so you can skip any monitoring code entirely if you don’t want to use this feature.

A registration example might look like:

This is the list of all monitoring events that you can register hooks for. The keys listed are present in the map that is passed to the callback function. The names of the events should readily identify what has taken place to trigger the callback.

Zookeeper tends to not get a huge amount of traffic, so this probably won’t offer a huge performance boost. It’s helpful if you’re trying to make your processing latency as predictable as possible, though.

Messaging requires the UDP port to be open for port set :onyx.messaging/peer-port.

All peers require the ability to connect to the ZooKeeper instances over TCP.

In a masterless design, there is no single entity that assigns tasks to peers. Instead, peers need to contend for tasks to execute as jobs are submitted to Onyx. Conversely, as peers are added to the cluster, the peers must "shift" to distribute the workload across the cluster. Onyx ships out-of-the-box job and task allocation policies. End users can change the levels of fairness that each job gets with respect to cluster power. And remember, one virtual peer executes at most one task.

It’s often the case that a set of machines in your cluster are privileged in some way. Perhaps they are running special hardware, or they live in a specific data center, or they have a license to use a proprietary database. Sometimes, you’ll have Onyx jobs that require tasks to run on a predetermined set of machines. Tags are a feature that let peers denote "capabilities". Tasks may declare which tags peers must have in order to be selected to execute them.

Onyx uses an internal log to totally order all coordination events across nodes in the cluster. This log is maintained as a directory of sequentially ordered znodes in ZooKeeper. It’s often of interest to watch the events as they are written to the log. For instance, you may want to know when a particular task is completed, or when a peer joins or leaves the cluster. You can use the log subscriber to do just that.

The following is a complete example to pretty print all events as they are written to the log. You need to provide the ZooKeeper address, Onyx ID, and shared job scheduler in the peer config. The subscriber will automatically track recover from sequentially reading errors in the case that a garbage collection is triggered, deleting log entries in its path.

Some example output from a test, printing the log position, log entry content, and the replica as-of that log entry:

In order to implement a plugin, one or more protocols need to be implemented from the Pipeline Extensions API. Reader plugins will implement PipelineInput and Pipeline. Writer plugins will implement Pipeline. See the docstrings for instructions on implementation.

To help move past the boilerplate of creating new plugins, use Leiningen with onyx-plugin to generate a template.

Often virtual peers allocated to a task may need to coordinate with respect to allocating work. For example, a Kafka reader task may need to assign partitions to different peers on the same topic. The Onyx mechanism for coordinating peers is the log. The Onyx log is extensible by plugins, by implementing several extensions defmethods.

For example:

When modifying the replica, please assoc-in into replica under [:task-metadata job-id task-id], so that it will be cleaned up when the job is completed or killed.

This plugin is included with Onyx. You do not need to add it as a separate dependency.

In your peer boot-up namespace:

Onyx uses Timbre for logging.

By default, all Onyx output is logged to a file called onyx.log in the same directory as where the peer or coordinator jar is executing. The logging configuration can be overridden completely, see below.

Onyx’s default logging configuration writes all WARN and FATAL messages to standard out, also.

See the Examples Project for a set of self-contained demonstrations of specific Onyx functionality.

When your job’s tasks don’t start, you won’t see any messages in onyx.log on the peer that read something like:

Instead, the log would show peers starting, but not doing anything. This looks like something:

If you see something like the following repeatedly being written to onyx.log, you’re in the right place:

What’s happening here is that your job’s tasks have started execution. Peers have been assigned tasks, and are beginning to get things in order to perform work on them. Before a peer can truly begin running a task, it must do three things:

The results from the first step are a sequence of boolean values designating whether the peer is allowed to start this task. If any of the start-task? lifecycle calls return false, the peer sleeps for a short period to back off, then performs step 1 again. This process continues until all lifecycle calls return true. Read the Lifecycles chapter of this User Guide to learn why it’s useful to be repeatedly try and back-off from task launching.

The second step will only occur after step 1 successfully completes - that is, all lifecycle functions return true. The reason you see "Backing off and retrying…​" in the logs is because step 1 is being repeated again.

If you’re onyx.log shows messages that read as follows, your job’s tasks have successfully started:

If you suspect that messages are not being processed, it heavily depends on the input plugin that you’re using.

Message replay happens when a mesage enters Onyx from an input source, gets processed, and is seen again at a later point in time. Onyx replays messages for fault tolerance when it suspects that failure of some sort has occurred. You can read about how message replay is implemented, and why it is exists, in the Architecture chapter of this User Guide.

There are many reasons why a message may need to be replayed (every possible failure scenario), so we will limit our discussion to controlling replay frequency. See the performance tuning sections of this document for more context about what value is appropriate to set for the replay frequency.

Programs that begin healthy by processing messages and then stall are out typically indicative of user-level code problems. We outline a few common cases here.

java.io.IOException: No space left on device

This exception commonly occurs when running Onyx inside of a Docker container. Aeron requires more shared memory than the container allocates by default. You can solve this problem by starting your container with a larger amount of shared memory by specifying --shm-size on Docker >= 1.10.

This exception can occur when Aeron does not have enough shared memory. Increase the amount of shared memory that is set as described above.

org.apache.bookkeeper.bookie.BookieException$InvalidCookieException: Cookie

This exception occurs due to a bug in BookKeeper reconnection to ZooKeeper before it’s ephemeral node expires. We are currently surveying our own workarounds until this is patched, but for now the thing to do is to delete /tmp/bookkeeper_journal and /tmp/bookkeeper_ledger on the host. Restart the peer, and all will be well.

java.lang.IllegalStateException: aeron cnc file version not understood

This exception occurs when Aeron’s version is upgraded or downgraded between incompatible versions. The exception will also provide a path on the OS to some Aeron files. Shutdown the peer, delete that directory, then restart the peer.

Failed to connect to the Media Driver - is it currently running?

This message is thrown when the peer tries to start, but can’t engage Aeron in its local environment. Aeron can be run in embedded mode by switching :onyx.messaging.aeron/embedded-driver? to true, or by running it out of process on the peer machine, which is the recommended production setting. If you’re running it out of process, ensure that it didn’t go down when you encounter this message. You should run Aeron through a process monitoring tool such as monit when running it out of process.

uk.co.real_logic.aeron.driver.exceptions.ActiveDriverException: active driver detected

You have encountered the following exception:

This is because you have started your peer-group twice without shutting it down. Alternatively, you may be using :onyx.messaging.aeron/embedded-driver? true in your peer-group and starting a media driver externally. Only one media driver can be started at a time.

'java.lang.unsupporteclassversionerror: uk.co.real_logic/aeron/Aeron$context unsupported major.minor version 52.0'

You have encountered the following exception:

This is because you are trying to build/run an Onyx app with a JRE version lower than 1.8. Onyx supports Java 1.8 only.

org.apache.bookkeeper.proto.WriteEntryProcessorV3: Error writing entry:X to ledger:Y

You have encountered the following exception:

Your ZooKeeper directory has been cleared out of information that points to the BookKeeper servers, and the two processes can’t sync up. This can be fixed by removing the data directory from the BookKeeper servers and ZooKeeper servers.

No implementation of method: :read-char of protocol: ='clojure.tools.reader.reader-types/Reader found for class

You’ll encounter this exception when your :onyx/fn returns something that is not EDN and Nippy serializable, which is required to send it over the network. Ensure that return values from :onyx/fn return either a map, or a vector of maps. All values within must be EDN serializable.

Unless otherwise overridden in the Peer Pipeline API, Onyx will use Nippy. This can be override by setting the peer configuration with :onyx.messaging/compress-fn and :onyx.messaging/decompress-fn. See the Information Model documentation for more information.

Use [Flow Conditions](\{\{ "/flow-conditions.html" | prepend: page.dir | prepend: site.baseurl }}) or return an empty vector from your :onyx/fn.

Return a vector of maps from :onyx/fn instead of a map. All maps at the top level of the vector will be unrolled and pushed downstream.

logs?

You should monitor these, however KeeperErrorCode = NodeExists are probably fine:

This is a peer just trying to recreate a ZooKeeper path that was already created by another peer, and it can be safely ignored.

Definitely turn off messaging short circuiting, as messaging short circuiting will improve performance in a way that is unrealistic for multi-node use. Remember to turn messaging short circuiting back on for production use, as it does improve performance overall.