Onyx

A masterless, fault-tolerant, high-performance distributed computation system for Clojure.

This is a community-maintained fork of onyx-platform/onyx (now archived), updated to run on modern JDKs.

Installation

Add to your project.clj:

[com.github.clojure-finance/onyx "0.17.2"]

Clojars also lists 0.1.0 and 0.2.0 from the fork's earlier version scheme. 0.2.0 has the same content as 0.15.0; prefer 0.15.0 or later.

JVM Requirements

JDK 17 or newer is required (tested on JDK 21). Onyx uses Aeron 1.53.x for messaging, and the Agrona library bundled with it reads jdk.internal.misc.Unsafe, so the JVM that starts peers needs this flag:

--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED

In Leiningen:

:jvm-opts ["--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED"]

Without this flag, the embedded media driver fails with:

IllegalAccessError: class org.agrona.UnsafeApi ... cannot access class jdk.internal.misc.Unsafe

On JDK 24 and newer, also pass --enable-native-access=ALL-UNNAMED. Otherwise the JVM warns that nippy's LZ4 compression calls a restricted method (System.load) that will be blocked in a future release. The flag exists since JDK 17, so it is safe to add unconditionally.

What is Onyx?

Onyx is a distributed computation system written in pure Clojure. It features:

• Masterless architecture — no single point of failure
• Cloud-scale fault tolerance and high performance
• Hybrid batch/stream processing model
• Information model for describing and constructing distributed workflows

It occupies the same space as Storm, Flink, Spark and similar systems.

Use Cases

• Real-time event stream processing
• Continuous computation and CQRS
• Extract, transform, load (ETL) pipelines
• Data transformation (map-reduce style)
• Data ingestion between storage systems
• Data cleaning and normalization

Changes in This Fork

0.17.2 (Latest)

• Fix: a windowed task after another windowed task never included what the first one's triggers emitted when the job completed in its own end-of-job results. A task with triggers now passes completion on one barrier after it seals, and the coordinator completes the job only once every task has sealed.
• The coordination log version is now 0.17.2. Use a new :onyx/tenancy-id when upgrading.

0.17.1

• Fix: segments that triggers emit when a job completes (such as the final windows) could be lost without an error. The coordinator now completes a job only after one more barrier has gone through, which each task passes on only after writing them.
• Reverts the change 0.17.0 made for this, which did not prevent the loss and kept the tasks after a task with triggers from sealing
• The coordination log version is now 0.17.1. Use a new :onyx/tenancy-id when upgrading.

0.17.0

• onyx.plugin.seq accepts a function for :seq/seq, so a large input need not fit in memory; onyx.tasks.seq/input-file uses this and now honours its task name
• Experimental onyx.replay (in-process runtime) and onyx.pit (point-in-time as-of joins), see below
• The coordination log version is now 0.17.0. Use a new :onyx/tenancy-id when upgrading.

0.16.0

• ZooKeeper 3.9.5, Curator 5.9.0, slf4j 1.7.36
• ZooKeeper now logs via slf4j. Onyx ships slf4j-nop, so add your own binding to see its logs.

0.15.0

• Aeron upgraded from 1.21.2 to 1.53.1
• JDK 17+ required (the previous sun.nio.ch and java.lang opens are no longer needed)
• The coordination log version is now 0.15.0. It had stayed at 0.14.6-SNAPSHOT, which silently skipped the cluster version compatibility check, so peers of different Onyx versions now refuse to share a tenancy. Use a new :onyx/tenancy-id when upgrading.

Earlier fork work (2023, before 0.15.0)

• Clojure 1.11, core.async 1.6.673, ZooKeeper 3.4.12 → 3.5.5, Curator 4.0.1 → 5.5.0
• onyx.static.uuid/random-uuid renamed to onyx-random-uuid to avoid clashing with clojure.core/random-uuid (Clojure 1.11)

See changes.md for the full changelog including upstream history.

Experimental: In-Process Runtime and Point-in-Time Joins

Added in 0.17.0. These namespaces may change in later releases.

onyx.replay runs a job map, the same one you would pass to onyx.api/submit-job, inside the calling JVM: no ZooKeeper, no Aeron, no JVM flags.

(require '[onyx.replay :as replay])

(replay/run-job job {:parallelism 4})
;; => {:segments {:in 1000, :inc 1000, :out 1000}}  ; segments each task emitted

• It blocks until every input is exhausted and every output has written everything.
• Output order is deterministic: task functions run on a thread pool, but results are emitted in input order.
• It supports plugins, task functions, :lifecycle/before-task-start and :lifecycle/after-task-stop, windows, triggers, group-by and watermarks. Watermarks travel with the batches, so watermark triggers fire at the same point in the data on every run.
• It rejects flow conditions, Java tasks and other lifecycle phases. There is no checkpointing: an exception fails the run.

onyx.pit builds point-in-time as-of joins from windows, triggers and refinements, so they run in onyx.replay and on a cluster. Segments carry :pit/known, the time their information became available (a filing date, say); right-hand segments may also carry :pit/event, the time it is about (a fiscal period end). Each left segment comes out with :pit/as-of set to the right segment with the same key and the latest :pit/event among those known by the left segment's :pit/known, or nil if none was.

(require '[onyx.pit :as pit])

(let [{:keys [catalog-entry window trigger]} (pit/as-of-join :join :tic 100)]
  (pit/check-job
   {:workflow [[:prices :join] [:filings :join] [:join :out]]
    :catalog [(pit/input prices-task :left)
              (pit/input filings-task :right)
              catalog-entry
              out-task]
    :windows [window]
    :triggers [trigger]
    :lifecycles lifecycles
    :task-scheduler :onyx.task-scheduler/balanced}))

Each input must be ordered by :pit/known; a right segment that arrives after results it would have changed have been emitted makes the join throw. See the onyx.pit docstring for the exact rules.

Documentation

Note: The upstream project is archived, so some external links may be outdated.

• User Guide — comprehensive documentation (sources in doc/user-guide)
• API reference — generated HTML is in doc/api; the public API lives in src/onyx/api.clj
• Data model reference — every job, catalog, lifecycle and peer-config key is described in src/onyx/information_model.cljc
• Onyx Starter — quick start project with walkthrough
• Onyx Template — Leiningen application template

Plugins

Built into Onyx core:

• onyx-core-async
• onyx-seq

Official plugins from upstream (may require updates for this fork):

• onyx-kafka
• onyx-datomic
• onyx-redis
• onyx-sql
• onyx-amazon-sqs
• onyx-amazon-s3
• onyx-http

Plugin template for building your own: onyx-plugin

Building from Source

Install into your local Maven repository:

lein install

Run the test suite:

lein test

The default selector skips tests tagged ^:broken or ^:stress. Run namespaces one at a time if you encounter issues — embedded ZooKeeper and the Aeron directory under /dev/shm can collide across parallel runs.

Contributors

Onyx was created by Michael Drogalis and developed by the upstream contributors.

Acknowledgements

Some code incorporated from:

• Riemann
• zookeeper-clj

License

Copyright © 2017 Michael Drogalis

Distributed under the Eclipse Public License, the same as Clojure.