diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 54f1387..26bfc6d 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -13,19 +13,19 @@ jobs:
matrix:
java: ["17", "21", "25"]
steps:
- - uses: actions/checkout@v4
- - uses: actions/setup-java@v5
+ - uses: actions/checkout@v7
+ - uses: actions/setup-java@v6
with:
distribution: temurin
java-version: ${{ matrix.java }}
- uses: DeLaGuardo/setup-clojure@13.7.0
with:
+ # 2.13.0's bin/lein still says LEIN_VERSION 2.12.1-SNAPSHOT, so
+ # setup-clojure cannot install it
lein: 2.12.0
- - uses: actions/cache@v4
+ - uses: actions/cache@v6
with:
path: ~/.m2/repository
key: m2-${{ runner.os }}-${{ hashFiles('project.clj') }}
restore-keys: m2-${{ runner.os }}-
- # Embedded ZooKeeper binds port 2188 and Aeron uses /dev/shm, so the
- # suite must run in a single JVM; lein test already does that.
- run: lein test
diff --git a/.gitignore b/.gitignore
index 9cc6d53..642f50a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,5 +1,4 @@
.DS_Store
-onyx.log*
*.log
resources/
outputs/
diff --git a/README.md b/README.md
index 699afd0..7e03d07 100644
--- a/README.md
+++ b/README.md
@@ -20,16 +20,14 @@ A Clojure dataframe library for **larger-than-memory** datasets. Process million
**Add to project.clj:**
```clojure
-[com.github.clojure-finance/clojask "2.0.6"]
+[com.github.clojure-finance/clojask "2.1.0"]
```
**Or deps.edn:**
```clojure
-com.github.clojure-finance/clojask {:mvn/version "2.0.6"}
+com.github.clojure-finance/clojask {:mvn/version "2.1.0"}
```
-Both also need two JVM flags, see [Requirements](#requirements).
-
**Basic example** (using [Employees-example.csv](test/clojask/Employees-example.csv)):
```clojure
(require '[clojask.dataframe :as ck])
@@ -90,44 +88,7 @@ Both also need two JVM flags, see [Requirements](#requirements).
- **OS:** macOS or Linux
- **JDK:** 17 or newer (tested on 17, 21, and 25)
-**Required JVM flags:**
-```clojure
-;; project.clj
-:jvm-opts ["--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED"
- "--enable-native-access=ALL-UNNAMED"]
-
-;; deps.edn, inside an alias (e.g. :aliases {:dev {...}}, run with -A:dev)
-:jvm-opts ["--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED"
- "--enable-native-access=ALL-UNNAMED"]
-```
-
-When running plain `java`, pass both flags on the command line.
-
-The first flag is required for Agrona (used by Onyx's messaging). Without it the first `compute` fails with `IllegalAccessError: class org.agrona.UnsafeApi ... cannot access class jdk.internal.misc.Unsafe`. The second is for lz4's native code: JDK 24+ prints a warning without it, and a future JDK will block the load.
-
-
-Running multiple Clojask processes
-
-Each `compute` starts embedded ZooKeeper (port 2188) and Aeron (port 40200), with Aeron's files in the default media-driver directory (`/dev/shm` on Linux). To run multiple processes on one machine, give each different settings:
-
-**System properties:**
-- `clojask.zookeeper.port`
-- `clojask.aeron.port`
-- `clojask.aeron.dir`
-
-**Or environment variables:**
-- `CLOJASK_ZOOKEEPER_PORT`
-- `CLOJASK_AERON_PORT`
-- `CLOJASK_AERON_DIR`
-
-
-
-
-Logging
-
-Onyx writes warnings and errors to `.clojask/clojask.log` in the working directory, rotated at 10 MB with one backup.
-
-
+Upgrading from 2.0.x: clojask no longer uses Onyx, so the JVM flags it needed (`--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED`, `--enable-native-access=ALL-UNNAMED`) and the `clojask.zookeeper.*` / `clojask.aeron.*` / `CLOJASK_*` port and directory settings can be removed.
## Documentation
@@ -138,7 +99,7 @@ Onyx writes warnings and errors to `.clojask/clojask.log` in the working directo
## How It Works
-Clojask uses [Onyx](https://github.com/clojure-finance/onyx) (a maintained fork) as a single-machine thread pool. Operations are collected lazily and executed together when you call `compute`, streaming data through your transformation pipeline without loading the full dataset into memory.
+Operations are collected lazily and executed together when you call `compute`, streaming data through your transformation pipeline without loading the full dataset into memory. Each `compute` runs in your JVM: one thread reads the input in batches, a pool of worker threads (the number you pass to `compute`) applies the operations and filters, and the calling thread writes the results in input order. Group-by and joins first sort the rows into groups by key, kept in temporary files under `.clojask/` (or in memory with `:in-memory true`); group-by and outer joins write their results in group order.
## Issues & Feedback
diff --git a/docs/documentation.md b/docs/documentation.md
index 0c4d6d6..e8275ae 100644
--- a/docs/documentation.md
+++ b/docs/documentation.md
@@ -444,7 +444,7 @@ This means you cannot further apply complicated operations to a joined dataframe
#### enable-debug / disable-debug
-Turn debug mode on or off. With debug mode on, operations are no longer checked against a preview when they are added, and logging is raised to `:debug`. Both take no arguments and print the new mode.
+Turn debug mode on or off. With debug mode on, operations are no longer checked against a preview when they are added. Both take no arguments and print the new mode.
```clojure
(enable-debug)
@@ -460,10 +460,10 @@ Compute the result. The pre-defined lazy operations will be executed in pipeline
| Argument | Type | Function | Remarks |
| ------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------------ |
| `dataframe` | clojask.classes.DataFrame.DataFrame / clojask.classes.JoinedDataFrame.JoinedDataFrame | The operated object | |
-| `num of workers` | int | The number of worker instances (except the input and output nodes) | Uses [Onyx](https://github.com/clojure-finance/onyx) as the distributed platform. Group-by and aggregate computes accept at most 8 workers. |
+| `num of workers` | int | The number of worker threads (besides the thread that reads the input and the calling thread, which writes the output) | An integer from 1 to 8, for every kind of compute. |
| `output path` | String / `nil` | The path of the output csv file | If the path already exists, will overwrite the file.
If `nil`, will store the output in memory as a vector of vectors, which represent each row, with the header row first unless `:header false`. See [example](https://github.com/clojure-finance/clojask-examples/blob/main/src/clojask_examples/in_memory.clj). |
| [`exception`] | Boolean | Whether an exception during calculation will cause termination | By default `false`. Is useful for debugging or detecting empty fields |
-| [`order`] | Boolean | If enforce the order of rows in the output to be the same as input | By default `false`. If set to `true`, will sacrifice the performance. |
+| [`order`] | Boolean | Has no effect since 2.1 | Accepted so that existing calls keep working. Output rows are always in input order, except that group-by and outer joins write theirs in group order. |
| [`output`] | Function | Specify how to output a row vector to the output file | Takes two arguments.
`writer` java.io.BufferedWriter
`rows` clojure.lang.PersistentVector (rows) of clojure.lang.PersistentVector (each row) |
| [`select`] | String / Collection of strings | Chooses columns to select for the operation | Can only specify either of select and exclude |
| [`exclude`] | String / Collection of strings | Chooses columns to be excluded for the operation | Can only specify either of select and exclude |
@@ -484,9 +484,6 @@ A `clojask.classes.DataFrame.DataFrame` read from the output file (read without
(compute x 8 "output.csv" :select "col a")
;; only select column a
-(compute x 8 "output.csv" :order true)
-;; make sure the order of the output is the same of the input
-
(compute x 8 nil :in-memory true)
;; compute the dataframe in memory and store the dataframe also in memory
diff --git a/project.clj b/project.clj
index 0f558b3..48fce74 100644
--- a/project.clj
+++ b/project.clj
@@ -1,17 +1,13 @@
-(defproject com.github.clojure-finance/clojask "2.0.6"
+(defproject com.github.clojure-finance/clojask "2.1.0"
:description "Data analysis and manipulation library with parallel computing for larger-than-memory datasets"
:url "https://github.com/clojure-finance/clojask"
:license {:name "MIT"
:url "https://github.com/clojure-finance/clojask/blob/1.x.x/LICENSE"}
- :dependencies [[org.clojure/clojure "1.11.1"]
+ :dependencies [[org.clojure/clojure "1.12.6"]
;; [org.clojure/math.numeric-tower "0.0.4"]
- [org.clojure/data.csv "1.0.0"]
- [com.github.clojure-finance/onyx "0.16.0"]
- [com.taoensso/timbre "5.2.1"]
;; [techascent/tech.ml.dataset "5.17" :exclusions [[ch.qos.logback/logback-classic][org.slf4j/slf4j-api]]]
- [com.google.code.externalsortinginjava/externalsortinginjava "0.6.0"]
- [com.github.clojure-finance/clojask-io "1.0.8"]
- [com.github.clojure-finance/clojure-heap "1.0.3"]]
+ [com.google.code.externalsortinginjava/externalsortinginjava "0.6.6"]
+ [com.github.clojure-finance/clojask-io "1.0.8"]]
;; lein deploy clojars, credentials from CLOJARS_USERNAME / CLOJARS_PASSWORD
:deploy-repositories [["clojars" {:url "https://clojars.org/repo"
:username :env/clojars_username
@@ -21,11 +17,7 @@
:source-paths ["src/main/clojure"]
:java-source-paths ["src/main/java"]
:javac-options ["--release" "17"]
- ;; Onyx's Aeron messaging (via Agrona) reads jdk.internal.misc.Unsafe.
- ;; lz4 (via Onyx's nippy serialization) calls System.load; JDK 24+ warns without
- ;; native access enabled.
- :jvm-opts ["-XX:+UseG1GC" "-server" "--add-opens=java.base/jdk.internal.misc=ALL-UNNAMED"
- "--enable-native-access=ALL-UNNAMED"]
+ :jvm-opts ["-XX:+UseG1GC" "-server"]
:test-paths ["test/clojask"]
;:java-test-paths ["test/java"]
;;:injections [(.. System (setProperty "clojure.core.async.pool-size" "8"))]
diff --git a/src/main/clojure/clojask/aggregate/aggre_input.clj b/src/main/clojure/clojask/aggregate/aggre_input.clj
deleted file mode 100644
index 55ff231..0000000
--- a/src/main/clojure/clojask/aggregate/aggre_input.clj
+++ /dev/null
@@ -1,89 +0,0 @@
-(ns clojask.aggregate.aggre-input
- (:require [onyx.plugin.protocols :as p]
- [taoensso.timbre :refer [info]]
- [clojure.java.io :as java.io]
- [clojask.utils :as u]))
-
-(defrecord AbsSeqReader [event path rst completed? checkpoint? offset source]
- p/Plugin
-
- (start [this event]
- this)
-
- (stop [this event]
- this)
-
- p/Checkpointed
- (checkpoint [this]
- (when checkpoint? @offset))
-
- (recover!
- [this _ checkpoint]
- (vreset! completed? false)
-
- (let [
- data (if (= path nil)
- (do
- (def tmp (volatile! -1))
- (map (fn [file]
- (vswap! tmp inc)
- {:id @tmp :file file :d (read-string file)})
- (.getKeys source)))
- (do
- (def tmp (volatile! -1))
- (map (fn [file]
- (vswap! tmp inc)
- {:id @tmp :file file :d (read-string (u/decode-str (.getName file)))})
- (rest (file-seq (java.io/file path))))))
- ]
- (if (nil? checkpoint)
- (do
- (vreset! rst data)
- (vreset! offset 0))
- (do
- (info "clojask.aggregate.aggre-input is recovering state by dropping" checkpoint "elements.")
- (vreset! rst (drop checkpoint data))
- (vreset! offset checkpoint)))))
-
- (checkpointed! [this epoch])
-
- p/BarrierSynchronization
- (synced? [this epoch]
- true)
-
- (completed? [this]
- @completed?)
-
- p/Input
- (poll! [this _ _]
- (if-let [seg (first @rst)]
- (do
- (vswap! rst rest)
- seg)
- (do (vreset! completed? true)
- nil))
- ))
-
-(defn inject-dataframe
- [dataframe _source]
- (def df dataframe)
- (def source _source))
-
-(defn input [{:keys [onyx.core/task-map] :as event}]
- (map->AbsSeqReader {:event event
- :path (:buffered-reader/path event)
- :rst (volatile! nil)
- :completed? (volatile! false)
- :checkpoint? (not (false? (:seq/checkpoint? task-map)))
- :offset (volatile! nil)
- :source source}))
-
-(def reader-calls
- {})
-
-(defn inject-lifecycle-seq
- [_ lifecycle]
- {:seq/seq (:seq/sequential lifecycle)})
-
-(def inject-seq-via-lifecycle
- {:lifecycle/before-task-start inject-lifecycle-seq})
\ No newline at end of file
diff --git a/src/main/clojure/clojask/aggregate/aggre_onyx_comps.clj b/src/main/clojure/clojask/aggregate/aggre_onyx_comps.clj
deleted file mode 100644
index b9b5721..0000000
--- a/src/main/clojure/clojask/aggregate/aggre_onyx_comps.clj
+++ /dev/null
@@ -1,150 +0,0 @@
-(ns clojask.aggregate.aggre-onyx-comps
- (:require [clojask.aggregate.aggre-input :as input]
- [clojask.aggregate.aggre-output :as output]
- [clojask.onyx-comps :as oc]
- [clojask.utils :as u]
- [clojask.groupby :refer [read-csv-seq]]))
-
-(def dataframe (atom nil))
-
-(defn worker-func-gen
- [df exception aggre-funcs index formatter source]
- (reset! dataframe df)
- (let [
- formatters formatter
- reorder (fn [a b]
- (u/gets (concat a b) index))
- groupby-keys (.getGroupbyKeys (:row-info df))
- groupby-index (mapv #(nth % 1) groupby-keys)
- org-format (u/formatters-by-position (.getFormatter (:col-info df)) groupby-index)
- pre-index (take (count groupby-index) (iterate inc 0))
- ]
- (defn worker-func
- "refered in preview"
- [seq]
- (let [data (if (= source nil) (read-csv-seq (:file seq)) (.getKey source (:file seq)))
- pre (:d seq)
- pre (u/gets-format pre pre-index org-format)
- data-map (-> (iterate inc 0)
- (zipmap (apply map vector data)))]
- (loop [aggre-funcs aggre-funcs
- res []]
- (if (= aggre-funcs [])
- (if (= res [])
- {:d [pre]}
- {:d (mapv reorder (repeat pre) (apply map vector res))})
- (let [func (first (first aggre-funcs))
- index (nth (first aggre-funcs) 1)
- res-funcs (rest aggre-funcs)
- new (func (get data-map index))
- new (if (coll? new)
- new
- (vector new))
- new (mapv (fn [_] (if-let [formatter (get formatters index)]
- (formatter _)
- (str _))) new)]
- (if (or (= res []) (= (count new) (count (last res))))
- (recur res-funcs (conj res new))
- (throw (Exception. "aggregation result is not of the same length")))
- )))
- ))))
-
-(defn catalog-gen
- "Generate the catalog for running Onyx"
- [num-work batch-size]
- ;; initialisation
- (def catalog [])
-
- ;; input
- (def catalog
- (conj catalog
- {:onyx/name :in
- :onyx/plugin :clojask.aggregate.aggre-input/input
- :onyx/type :input
- :onyx/medium :seq
- :seq/checkpoint? true
- :onyx/batch-size batch-size
- :onyx/max-peers 1
- :input/doc "Reads segments from a core.async channel"}))
-
- ;; for loop for sample workers
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))
- worker-function (keyword "clojask.aggregate.aggre-onyx-comps" "worker-func")]
- (def catalog
- (conj catalog
- {:onyx/name worker-name
- :onyx/fn worker-function
- :onyx/type :function
- :onyx/batch-size batch-size
- :worker/doc "This is a worker node"}
- ))))
-
- ;; output
- (def catalog
- (conj catalog
- {:onyx/name :output
- :onyx/plugin :clojask.aggregate.aggre-output/output
- :onyx/type :output
- :onyx/medium :core.async ;; this is maked up
- :onyx/max-peers 1
- :onyx/batch-size batch-size
- :output/doc "Writes segments to the file"}))
-
- )
-
-(defn inject-in-reader [event lifecycle]
- (let []
- {:buffered-reader/path (:buffered-reader/path lifecycle)
- }))
-
-(def in-calls
- {:lifecycle/before-task-start inject-in-reader})
-
-(defn lifecycle-gen
- [source dist]
- (def lifecycles
- [{:lifecycle/task :in
- :buffered-reader/path source
- :lifecycle/calls ::in-calls}
- {:lifecycle/task :in
- :lifecycle/calls :clojask.aggregate.aggre-input/reader-calls}
- {:lifecycle/task :output
- :buffered-wtr/filename dist
- :lifecycle/calls :clojask.aggregate.aggre-output/writer-calls}]))
-
-(defn flow-cond-gen
- "Generate the flow conditions for running Onyx"
- [num-work]
- (def flow-conditions []) ;; initialisation
-
- ;; for loop for sample workers
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))
- predicate-function (keyword "clojask.aggregate.aggre-onyx-comps" (str "rem" (- x 1) "?"))]
- (intern 'clojask.aggregate.aggre-onyx-comps (symbol (str "rem" (- x 1) "?")) (fn [event old-segment new-segment all-new-segment]
- (= (mod (:id new-segment) num-work) (- x 1))))
- (def flow-conditions
- (conj flow-conditions
- {:flow/from :in
- :flow/to [worker-name]
- :flow/predicate predicate-function
- :worker/doc "This is a flow condition"}
- ))))
-
- )
-
-(defn start-onyx-aggre
- "start the onyx cluster with the specification inside dataframe"
- [num-work batch-size dataframe source dist exception aggre-func index formatter out]
- (oc/with-onyx-env "groupby aggregate" num-work
- (fn []
- (oc/run-job "groupby aggregate"
- (fn []
- (worker-func-gen dataframe exception aggre-func index formatter source)
- (catalog-gen num-work batch-size)
- (lifecycle-gen (if (nil? source) "./.clojask/grouped" nil) dist)
- (flow-cond-gen num-work)
- (input/inject-dataframe dataframe source)
- (output/inject-dataframe dataframe out)
- {:catalog catalog :lifecycles lifecycles :flow-conditions flow-conditions})))))
\ No newline at end of file
diff --git a/src/main/clojure/clojask/aggregate/aggre_output.clj b/src/main/clojure/clojask/aggregate/aggre_output.clj
deleted file mode 100644
index ec118d2..0000000
--- a/src/main/clojure/clojask/aggregate/aggre_output.clj
+++ /dev/null
@@ -1,87 +0,0 @@
-(ns clojask.aggregate.aggre-output
- (:require [onyx.plugin.protocols :as p]
- [clojure.java.io :as io]))
-
-(defn- inject-into-eventmap
- [event lifecycle]
- (let [wtr (io/writer (:buffered-wtr/filename lifecycle) :append true)]
- {:clojask/wtr wtr}))
-
-(defn- close-writer [event lifecycle]
- (.close (:clojask/wtr event)))
-
-;; Map of lifecycle calls that are required to use this plugin.
-;; Users will generally always have to include these in their lifecycle calls
-;; when submitting the job.
-(def writer-calls
- {:lifecycle/before-task-start inject-into-eventmap
- :lifecycle/after-task-stop close-writer})
-
-(def df (atom nil))
-(def output-func (atom nil))
-
-(defn inject-dataframe
- [dataframe out]
- (reset! df dataframe)
- (reset! output-func out))
-
-(defrecord ClojaskOutput [output-func]
- p/Plugin
- (start [this event]
- ;; Initialize the plugin, generally by assoc'ing any initial state.
- this)
-
- (stop [this event]
- ;; Nothing is required here. However, most plugins have resources
- ;; (e.g. a connection) to clean up.
- ;; Mind that such cleanup is also achievable with lifecycles.
- this)
-
- p/Checkpointed
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpoint [this])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (recover! [this replica-version checkpoint])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpointed! [this epoch])
-
- p/BarrierSynchronization
- (synced? [this epoch]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished.
- true)
-
- (completed? [this]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished (just like synced).
- true)
-
- p/Output
- (prepare-batch [this event replica messenger]
- ;; Nothing is required here. This is useful for some initial preparation,
- ;; before write-batch is called repeatedly.
- true)
-
- (write-batch [this {:keys [onyx.core/write-batch clojask/wtr]} replica messenger]
- (doseq [msg write-batch]
- (do
- (if (not= (:d msg) nil)
- (do
- (output-func wtr (:d msg))
- ;; !! define argument (debug)
- ))))
- (.flush wtr)
- true))
-
-;; Builder function for your output plugin.
-;; Instantiates a record.
-;; It is highly recommended you inject and pre-calculate frequently used data
-;; from your task-map here, in order to improve the performance of your plugin
-;; Extending the function below is likely good for most use cases.
-(defn output [pipeline-data]
- (->ClojaskOutput (deref output-func)))
\ No newline at end of file
diff --git a/src/main/clojure/clojask/classes/DataFrame.clj b/src/main/clojure/clojask/classes/DataFrame.clj
index 2aafabb..2931d72 100644
--- a/src/main/clojure/clojask/classes/DataFrame.clj
+++ b/src/main/clojure/clojask/classes/DataFrame.clj
@@ -6,13 +6,9 @@
[clojure.java.io :as io]
[clojask.utils :as u]
[clojask.engine :as engine]
- [clojask.onyx-comps :refer [start-onyx start-onyx-aggre-only start-onyx-groupby]]
[clojask.sort :as sort]
- [clojask.aggregate.aggre-onyx-comps :refer [start-onyx-aggre]]
- [clojask.join.outer-onyx-comps]
[clojure.string :as str]
- [clojask.preview :as preview]
- [clojask.join.outer-output])
+ [clojask.preview :as preview])
(:import [clojask.classes.ColInfo ColInfo]
[clojask.classes.RowInfo RowInfo]
[clojask.classes.DataStat DataStat]
@@ -127,9 +123,7 @@
index (if (= select [nil]) (take (count key-index) (iterate inc 0)) (keep key-index select))]
(assert (or (= (count select) (count index)) (= select [nil])) "Must select existing columns; get-col-names lists them.")
(if (= ifheader true) (.printCol this output-dir index out))
- (let [res (if (= (engine/engine) :local)
- (engine/start-compute num-worker batch-size this output-dir exception index melt out)
- (start-onyx num-worker batch-size this output-dir exception order index melt out))]
+ (let [res (engine/start-compute num-worker batch-size this output-dir exception index melt out)]
(if (= res "success")
"success"
"failed"))))
@@ -318,9 +312,7 @@
(.indexOf index num))])
aggre-func (mapv shift-func aggre-func)
tmp (if (= ifheader true) (.printCol this output-dir select out))
- res (if (= (engine/engine) :local)
- (engine/start-aggre-only num-worker batch-size this output-dir exception aggre-func index select out)
- (start-onyx-aggre-only num-worker batch-size this output-dir exception aggre-func index select out))]
+ res (engine/start-aggre-only num-worker batch-size this output-dir exception aggre-func index select out)]
(if (= res "success")
"success"
"failed")))
@@ -338,8 +330,7 @@
data-index (mapv #(- % (count groupby-keys)) (remove #(< % (count groupby-keys)) select))
groupby-index (vec (apply sorted-set (mapv #(nth % 1) (concat groupby-keys (u/gets aggre-keys data-index)))))
mgroup (MGroup. (transient {}))
- local? (= (engine/engine) :local)
- res ((if local? engine/start-groupby start-onyx-groupby)
+ res (engine/start-groupby
num-worker batch-size this (if inmemory mgroup ".clojask/grouped/") groupby-keys groupby-index exception)]
(if (= aggre-keys [])
(println (str "Since the dataframe is only grouped by but not aggregated, the result will be the same as to choose the distinct values of "
@@ -353,7 +344,7 @@
formatter (.getFormatter (.col-info this))
formatter (u/formatters-by-position formatter groupby-index)]
(.final mgroup)
- (if (= "success" ((if local? engine/start-aggre start-onyx-aggre)
+ (if (= "success" (engine/start-aggre
num-worker batch-size this (if inmemory mgroup nil) output-dir exception aggre-func select formatter out))
"success"
(throw (OperationException. "Error when aggregating."))))
@@ -377,23 +368,23 @@
(recur res true)
false)))))))
(throw (TypeException. "The order list is not in the correct format.")))
- (defn clojask-compare
- "a and b are two rows"
- [a b]
- (loop [list list key false sign +]
- (if (= list '())
- 0
- (let [_ (first list)
- res (rest list)]
- (if key
- (let [tmp (compare (u/get-key a (.getType col-info) (.getKeyIndex col-info) _) (u/get-key b (.getType col-info) (.getKeyIndex col-info) _))]
- (if (not= tmp 0)
- (sign tmp)
- (recur res false nil)))
- (if (= _ "+")
- (recur res true +)
- (recur res true -)))))))
- (sort/use-external-sort path output-dir clojask-compare))
+ (let [;; a and b are two rows
+ clojask-compare
+ (fn [a b]
+ (loop [list list key false sign +]
+ (if (= list '())
+ 0
+ (let [_ (first list)
+ res (rest list)]
+ (if key
+ (let [tmp (compare (u/get-key a (.getType col-info) (.getKeyIndex col-info) _) (u/get-key b (.getType col-info) (.getKeyIndex col-info) _))]
+ (if (not= tmp 0)
+ (sign tmp)
+ (recur res false nil)))
+ (if (= _ "+")
+ (recur res true +)
+ (recur res true -)))))))]
+ (sort/use-external-sort path output-dir clojask-compare)))
(previewDF
[this]
diff --git a/src/main/clojure/clojask/classes/JoinedDataFrame.clj b/src/main/clojure/clojask/classes/JoinedDataFrame.clj
index f6a9ca1..ec8cc84 100644
--- a/src/main/clojure/clojask/classes/JoinedDataFrame.clj
+++ b/src/main/clojure/clojask/classes/JoinedDataFrame.clj
@@ -6,8 +6,6 @@
[clojask.classes.MGroup]
[clojask.classes.DataFrame]
[clojask.engine :as engine]
- [clojask.onyx-comps :refer [start-onyx-groupby start-onyx-join]]
- [clojask.join.outer-onyx-comps :refer [start-onyx-outer]]
[clojure.java.io :as io]
[clojask.utils])
(:import [clojask.classes.MGroup MGroupJoin MGroupJoinOuter]
@@ -106,11 +104,9 @@
write-index (mapv (fn [num] (count (remove #(>= % num) (concat a-index (mapv #(+ % na) b-index))))) internal)
mgroup-a (MGroupJoinOuter. (transient {}) (transient {}) false)
mgroup-b (if (not= type 3) (MGroupJoin. (transient {}) (transient {}) (or (= 4 type) (= 5 type))) (MGroupJoinOuter. (transient {}) (transient {}) (or (= 4 type) (= 5 type))))
- local? (= (engine/engine) :local)
- ;; Onyx takes segments per batch, the local engine rows per batch
- group (fn [df & args] (apply (if local? engine/start-groupby start-onyx-groupby) num-worker (if local? (:batch-size df) 10) df args))
- join (fn [& args] (apply (if local? engine/start-join start-onyx-join) num-worker (if local? (:batch-size a) 10) a args))
- outer (fn [& args] (apply (if local? engine/start-outer start-onyx-outer) num-worker 10 a args))]
+ group (fn [df & args] (apply engine/start-groupby num-worker (:batch-size df) df args))
+ join (fn [& args] (apply engine/start-join num-worker (:batch-size a) a args))
+ outer (fn [& args] (apply engine/start-outer num-worker (:batch-size a) a args))]
(if (= ifheader true) (.printCol this output-dir select out))
(try
(cond
diff --git a/src/main/clojure/clojask/clojask_aggre.clj b/src/main/clojure/clojask/clojask_aggre.clj
deleted file mode 100644
index 842cc71..0000000
--- a/src/main/clojure/clojask/clojask_aggre.clj
+++ /dev/null
@@ -1,124 +0,0 @@
-(ns clojask.clojask-aggre
- (:require [onyx.plugin.protocols :as p]
- [clojure.java.io :as io]
- [clojask.api.aggregate :refer [start]]
- [clojask.utils :as u])
- (:import [com.clojask.exception ExecutionException]))
-
-(def df (atom nil))
-(def aggre-func (atom nil))
-(def select (atom nil))
-(def output-func (atom nil))
-
-(defn inject-dataframe
- [dataframe a b out]
- (reset! df dataframe)
- (reset! aggre-func a)
- (reset! select b)
- (reset! output-func out)
- )
-
-(defn c-count
- [a]
- (if (coll? a)
- (count a)
- 1))
-
-(defn- inject-into-eventmap
- [event lifecycle]
- (let [wtr (io/writer (:buffered-wtr/filename lifecycle) :append true)
- order (:order lifecycle)
- aggre-func (.getAggreFunc (:row-info (deref df)))]
- {:clojask/wtr wtr
- :clojask/aggre-func aggre-func}))
-
-(defn- close-writer [event lifecycle]
- (.close (:clojask/wtr event)))
-
-;; Map of lifecycle calls that are required to use this plugin.
-;; Users will generally always have to include these in their lifecycle calls
-;; when submitting the job.
-(def writer-calls
- {:lifecycle/before-task-start inject-into-eventmap
- :lifecycle/after-task-stop close-writer})
-
-(defrecord ClojaskOutput
- [memo
- aggre-func
- select
- output-func]
- p/Plugin
- (start [this event]
- ;; Initialize the plugin, generally by assoc'ing any initial state.
- this)
-
- (stop [this event]
- ;; Nothing is required here. However, most plugins have resources
- ;; (e.g. a connection) to clean up.
- ;; Mind that such cleanup is also achievable with lifecycles.
- ;; a memo still holding the start sentinel saw no rows: write an empty cell
- (let [data (mapv (fn [_] (cond (identical? _ start) [nil] (coll? _) _ :else [_])) (deref memo))
- wtr (:clojask/wtr event)]
- (if (apply = (map count data))
- (do
- (mapv
- ;; #(.write (:clojask/wtr event) (str (string/join "," (u/gets % select)) "\n"))
- (fn [msg] (output-func wtr [(u/gets msg select)])) (apply map vector data))
- (.flush wtr))
- (throw (ExecutionException. "aggregation result is not of the same length"))))
- this)
-
- p/Checkpointed
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpoint [this])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (recover! [this replica-version checkpoint])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpointed! [this epoch])
-
- p/BarrierSynchronization
- (synced? [this epoch]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished.
- true)
-
- (completed? [this]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished (just like synced).
- true)
-
- p/Output
- (prepare-batch [this event replica messenger]
- ;; Nothing is required here. This is useful for some initial preparation,
- ;; before write-batch is called repeatedly.
- true)
-
- (write-batch [this {:keys [onyx.core/write-batch clojask/wtr]} replica messenger]
- (let []
- (doseq [msg write-batch]
- (doseq [data (:d msg)]
- (if (not= data nil)
- (let [
- ]
-
- (vreset! memo (doall (map-indexed (fn [ind prev] ((nth (nth aggre-func ind) 0) prev (nth data (nth (nth aggre-func ind) 1)))) (deref memo))))
- )))))
- true))
-
-;; Builder function for your output plugin.
-;; Instantiates a record.
-;; It is highly recommended you inject and pre-calculate frequently used data
-;; from your task-map here, in order to improve the performance of your plugin
-;; Extending the function below is likely good for most use cases.
-(defn output [pipeline-data]
- (let []
- (->ClojaskOutput (volatile! (doall (take (count (deref aggre-func))
- (repeat start))))
- (deref aggre-func)
- (deref select)
- (deref output-func))))
\ No newline at end of file
diff --git a/src/main/clojure/clojask/clojask_groupby.clj b/src/main/clojure/clojask/clojask_groupby.clj
deleted file mode 100644
index bca5071..0000000
--- a/src/main/clojure/clojask/clojask_groupby.clj
+++ /dev/null
@@ -1,93 +0,0 @@
-(ns clojask.clojask-groupby
- (:require [clojask.groupby :refer [output-groupby]]
- [onyx.plugin.protocols :as p]))
-
-(def dataframe (atom nil))
-(def groupby-keys (atom nil))
-(def write-index (atom nil))
-(def output-func (atom nil))
-(def dist (atom nil))
-(def format_ (atom nil))
-
-(defn inject-dataframe
- [df groupby-key index _dist _format]
- (reset! dataframe df)
- (reset! groupby-keys groupby-key)
- (reset! write-index index)
- (reset! dist _dist)
- (reset! format_ _format)
- )
-
-(defn- inject-into-eventmap
- [event lifecycle]
- (let [key-index (.getKeyIndex (.col-info (deref dataframe)))
- formatters (.getFormatter (.col-info (deref dataframe)))
- groupby-keys (deref groupby-keys)]
- {:clojask/dist (deref dist)
- :clojask/groupby-keys groupby-keys
- :clojask/key-index key-index
- :clojask/formatter formatters}))
-
-;; Map of lifecycle calls that are required to use this plugin.
-;; Users will generally always have to include these in their lifecycle calls
-;; when submitting the job.
-(def writer-aggre-calls
- {:lifecycle/before-task-start inject-into-eventmap})
-
-(defrecord ClojaskGroupby [write-index]
- p/Plugin
- (start [this event]
- ;; Initialize the plugin, generally by assoc'ing any initial state.
- this)
-
- (stop [this event]
- ;; Nothing is required here. However, most plugins have resources
- ;; (e.g. a connection) to clean up.
- ;; Mind that such cleanup is also achievable with lifecycles.
- this)
-
- p/Checkpointed
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpoint [this])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (recover! [this replica-version checkpoint])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpointed! [this epoch])
-
- p/BarrierSynchronization
- (synced? [this epoch]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished.
- true)
-
- (completed? [this]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished (just like synced).
- true)
-
- p/Output
- (prepare-batch [this event replica messenger]
- ;; Nothing is required here. This is useful for some initial preparation,
- ;; before write-batch is called repeatedly.
- true)
-
- (write-batch [this {:keys [onyx.core/write-batch clojask/dist clojask/groupby-keys clojask/key-index clojask/formatter]} replica messenger]
- (doseq [msg write-batch]
- (doseq [data (:d msg)]
- (if (not= data nil)
- (do
- (output-groupby dist data groupby-keys key-index formatter write-index (deref format_))))))
- true))
-
-;; Builder function for your output plugin.
-;; Instantiates a record.
-;; It is highly recommended you inject and pre-calculate frequently used data
-;; from your task-map here, in order to improve the performance of your plugin
-;; Extending the function below is likely good for most use cases.
-(defn groupby [pipeline-data]
- (->ClojaskGroupby (deref write-index)))
\ No newline at end of file
diff --git a/src/main/clojure/clojask/clojask_input.clj b/src/main/clojure/clojask/clojask_input.clj
deleted file mode 100644
index 15c10e2..0000000
--- a/src/main/clojure/clojask/clojask_input.clj
+++ /dev/null
@@ -1,76 +0,0 @@
-(ns clojask.clojask-input
- (:require [onyx.plugin.protocols :as p]
- [clojask.utils]
- [taoensso.timbre :refer [info]]))
-
-(defrecord AbsSeqReader [event reader filters types have-col rst completed? checkpoint? offset batch-size]
- p/Plugin
-
- (start [this event]
- this)
-
- (stop [this event]
- this)
-
- p/Checkpointed
- (checkpoint [this]
- (when checkpoint? @offset))
-
- (recover! [this _ checkpoint]
- (vreset! completed? false)
- (let [csv-data (reader)
- data (map zipmap (repeat [:id :d]) (map vector (iterate inc 0) (partition batch-size batch-size [] csv-data)))]
- (if (nil? checkpoint)
- (do
- (vreset! rst data)
- (vreset! offset 0))
- (do
- (info "clojask.clojask-input is recovering state by dropping" checkpoint "elements.")
- (vreset! rst (drop checkpoint data))
- (vreset! offset checkpoint)))))
-
- (checkpointed! [this epoch])
-
- p/BarrierSynchronization
- (synced? [this epoch]
- true)
-
- (completed? [this]
- @completed?)
-
- p/Input
- (poll! [this _ _]
- (if-let [seg (first @rst)]
- (do
- (vswap! rst rest)
- seg
- )
- (do (vreset! completed? true)
- nil))
- ))
-
-(defn inject-dataframe
- [dataframe]
- (def df dataframe))
-
-(defn input [{:keys [onyx.core/task-map] :as event}]
- (map->AbsSeqReader {:event event
- :reader (.getFunc df)
- :filters (.getFilters (:row-info df))
- :types (.getType (:col-info df))
- :have-col (:have-col df)
- :rst (volatile! nil)
- :completed? (volatile! false)
- :checkpoint? (not (false? (:seq/checkpoint? task-map)))
- :offset (volatile! nil)
- :batch-size (:batch-size df)}))
-
-(def reader-calls
- {})
-
-(defn inject-lifecycle-seq
- [_ lifecycle]
- {:seq/seq (:seq/sequential lifecycle)})
-
-(def inject-seq-via-lifecycle
- {:lifecycle/before-task-start inject-lifecycle-seq})
\ No newline at end of file
diff --git a/src/main/clojure/clojask/clojask_join.clj b/src/main/clojure/clojask/clojask_join.clj
deleted file mode 100644
index 9db8158..0000000
--- a/src/main/clojure/clojask/clojask_join.clj
+++ /dev/null
@@ -1,112 +0,0 @@
-(ns clojask.clojask-join
- (:require [clojask.join :as join]
- [clojure.java.io :as io]
- [onyx.plugin.protocols :as p]))
-
-(def a (atom nil))
-(def b (atom nil))
-(def a-keys (atom nil))
-(def b-keys (atom nil))
-(def a-index (atom nil))
-(def b-index (atom nil))
-(def b-format (atom nil))
-(def join-index (atom nil))
-(def output-func (atom nil))
-
-(defn inject-dataframe
- [d-a d-b a-key b-key -a-index -b-index -join-index -b-format out]
- (reset! a d-a)
- (reset! b d-b)
- (reset! a-keys a-key)
- (reset! b-keys b-key)
- (reset! a-index -a-index)
- (reset! b-index -b-index)
- (reset! b-format -b-format)
- (reset! join-index -join-index)
- (reset! output-func out))
-
-(defn- inject-into-eventmap
- [event lifecycle]
- (let [wtr (io/writer (:buffered-wtr/filename lifecycle) :append true)
- a-format (.getFormatter (.col-info (deref a)))
- b-format (deref b-format)
- ]
-
- {:clojask/wtr wtr
- :clojask/a-keys (deref a-keys)
- :clojask/b-keys (deref b-keys)
- :clojask/a-roll (:clojask/a-roll lifecycle)
- :clojask/b-roll (:clojask/b-roll lifecycle)
- :clojask/a-map (:clojask/a-map lifecycle)
- :clojask/b-map (:clojask/b-map lifecycle)
- :clojask/a-format a-format
- :clojask/b-format b-format
- :clojask/join-type (:clojask/join-type lifecycle)}))
-
-(defn- close-writer [event lifecycle]
- (.close (:clojask/wtr event)))
-
-;; Map of lifecycle calls that are required to use this plugin.
-;; Users will generally always have to include these in their lifecycle calls
-;; when submitting the job.
-(def writer-join-calls
- {:lifecycle/before-task-start inject-into-eventmap
- :lifecycle/after-task-stop close-writer})
-
-(defrecord ClojaskJoin [a-index b-index join-index write-func]
- p/Plugin
- (start [this event]
- ;; Initialize the plugin, generally by assoc'ing any initial state.
- this)
-
- (stop [this event]
- ;; Nothing is required here. However, most plugins have resources
- ;; (e.g. a connection) to clean up.
- ;; Mind that such cleanup is also achievable with lifecycles.
- this)
-
- p/Checkpointed
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpoint [this])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (recover! [this replica-version checkpoint])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpointed! [this epoch])
-
- p/BarrierSynchronization
- (synced? [this epoch]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished.
- true)
-
- (completed? [this]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished (just like synced).
- true)
-
- p/Output
- (prepare-batch [this event replica messenger]
- ;; Nothing is required here. This is useful for some initial preparation,
- ;; before write-batch is called repeatedly.
- true)
-
- (write-batch [this {:keys [onyx.core/write-batch clojask/wtr clojask/a-keys clojask/b-keys clojask/a-roll clojask/b-roll clojask/a-map clojask/b-map clojask/a-format clojask/b-format clojask/join]} replica messenger]
- (doseq [msg write-batch]
- (doseq [data (:d msg)]
- (if (not= data nil)
- (do
- (join/output-join wtr data a-keys a-map b-keys (count b-map) a-roll b-roll a-format b-format a-index b-index join-index write-func)))))
- true))
-
-;; Builder function for your output plugin.
-;; Instantiates a record.
-;; It is highly recommended you inject and pre-calculate frequently used data
-;; from your task-map here, in order to improve the performance of your plugin
-;; Extending the function below is likely good for most use cases.
-(defn join [pipeline-data]
- (->ClojaskJoin (deref a-index) (deref b-index) (deref join-index) (deref output-func)))
\ No newline at end of file
diff --git a/src/main/clojure/clojask/clojask_output.clj b/src/main/clojure/clojask/clojask_output.clj
deleted file mode 100644
index aaf4951..0000000
--- a/src/main/clojure/clojask/clojask_output.clj
+++ /dev/null
@@ -1,120 +0,0 @@
-(ns clojask.clojask-output
- (:require [onyx.plugin.protocols :as p]
- [clojure.java.io :as io]
- [clojure-heap.core :as heap]
- [clojask.join.outer-output]))
-
-(def df (atom nil))
-(def output-func (atom nil))
-
-(defn inject-dataframe
- [dataframe out]
- (reset! df dataframe)
- (reset! output-func out)
- )
-
-(defn- inject-into-eventmap
- [event lifecycle]
- (let [wtr (io/writer (:buffered-wtr/filename lifecycle) :append true)
- order (:order lifecycle)
- indices (:indices lifecycle)
- formatter (.getFormatter (:col-info (deref df)))]
- {:clojask/wtr wtr :clojask/order order
- }))
-
-(defn- close-writer [event lifecycle]
- (.close (:clojask/wtr event)))
-
-(defn- write-msg
- [wtr msg melt output-func]
- (doseq [row (remove nil? (:d msg))]
- (output-func wtr (melt row)))
- )
-
-(defn- order-write
- [wtr msg heap exp-id melt output-func]
- (let [id (:id msg)]
- (if (= id (deref exp-id))
- (do
- (write-msg wtr msg melt output-func)
- (swap! exp-id inc)
- (while (= (:id (heap/peek heap)) (deref exp-id))
- (write-msg wtr (heap/poll heap) melt output-func)
- (swap! exp-id inc)))
- (do
- (heap/add heap msg)
- )
- )))
-
-;; Map of lifecycle calls that are required to use this plugin.
-;; Users will generally always have to include these in their lifecycle calls
-;; when submitting the job.
-(def writer-calls
- {:lifecycle/before-task-start inject-into-eventmap
- :lifecycle/after-task-stop close-writer})
-
-(def melt (atom nil))
-
-(defn inject-melt
- [tmp]
- (reset! melt tmp))
-
-(defrecord ClojaskOutput [melt heap exp-id output]
- p/Plugin
- (start [this event]
- ;; Initialize the plugin, generally by assoc'ing any initial state.
- this)
-
- (stop [this event]
- ;; Nothing is required here. However, most plugins have resources
- ;; (e.g. a connection) to clean up.
- ;; Mind that such cleanup is also achievable with lifecycles.
- (if (not= (heap/get-size heap) 0) (throw (Exception. (str "The order enforcement failed. " (heap/get-size heap) " rows have been shuffled or missing."))))
- this)
-
- p/Checkpointed
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpoint [this])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (recover! [this replica-version checkpoint])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpointed! [this epoch])
-
- p/BarrierSynchronization
- (synced? [this epoch]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished.
- true)
-
- (completed? [this]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished (just like synced).
- true)
-
- p/Output
- (prepare-batch [this event replica messenger]
- ;; Nothing is required here. This is useful for some initial preparation,
- ;; before write-batch is called repeatedly.
- true)
-
- (write-batch [this {:keys [onyx.core/write-batch clojask/wtr clojask/order]} replica messenger]
- (if order
- (doseq [msg write-batch]
- (order-write wtr msg heap exp-id melt output))
- (let []
- (doseq [msg write-batch]
- (write-msg wtr msg melt output))))
- true))
-
-;; Builder function for your output plugin.
-;; Instantiates a record.
-;; It is highly recommended you inject and pre-calculate frequently used data
-;; from your task-map here, in order to improve the performance of your plugin
-;; Extending the function below is likely good for most use cases.
-(defn output [pipeline-data]
- (->ClojaskOutput (deref melt) (heap/heap (fn [a b] (<= (:id a) (:id b)))) (atom 0) (deref output-func)))
\ No newline at end of file
diff --git a/src/main/clojure/clojask/dataframe.clj b/src/main/clojure/clojask/dataframe.clj
index 6f7d51e..f70fc3d 100644
--- a/src/main/clojure/clojask/dataframe.clj
+++ b/src/main/clojure/clojask/dataframe.clj
@@ -11,8 +11,7 @@
[clojure.pprint :as pprint]
[clojure.set :as set]
[clojure.string :as str]
- [clojure.java.io :as io]
- [taoensso.timbre :as timbre])
+ [clojure.java.io :as io])
(:import [clojask.classes.ColInfo ColInfo]
[clojask.classes.RowInfo RowInfo]
[clojask.classes.DataFrame DataFrame]
@@ -22,18 +21,14 @@
;; debug APIs
-(timbre/merge-config! {:min-level :warn})
-
(defn enable-debug
[]
(reset! clojask.classes.DataFrame/debug true)
- (timbre/merge-config! {:min-level :debug})
(println "Debug mode is on."))
(defn disable-debug
[]
(reset! clojask.classes.DataFrame/debug false)
- (timbre/merge-config! {:min-level :warn})
(println "Debug mode is off."))
;; public APIs of the dataframe
diff --git a/src/main/clojure/clojask/engine.clj b/src/main/clojure/clojask/engine.clj
index bcf6ddb..adc7828 100644
--- a/src/main/clojure/clojask/engine.clj
+++ b/src/main/clojure/clojask/engine.clj
@@ -1,9 +1,7 @@
(ns clojask.engine
- "A thread-pool engine for the jobs clojask runs: one input, num-work
- identical workers and one output, all in this JVM. It replaces Onyx one
- driver at a time. The system property clojask.engine, else the
- environment variable CLOJASK_ENGINE, chooses the engine: \"onyx\" (the
- default) or \"local\". Drivers not ported yet always run on Onyx."
+ "The thread-pool engine that runs every compute: one input, num-work
+ identical workers and one output, all in this JVM. It replaced Onyx in
+ clojask 2.1."
(:require [clojure.java.io :as io]
[clojure.string :as string]
[clojask.api.aggregate :refer [start]]
@@ -14,15 +12,6 @@
[java.util.concurrent ArrayBlockingQueue Callable ExecutorService
Executors Future ThreadFactory TimeUnit]))
-(defn engine
- "The engine for the next compute, :onyx or :local."
- []
- (let [v (or (System/getProperty "clojask.engine") (System/getenv "CLOJASK_ENGINE") "onyx")]
- (case v
- "onyx" :onyx
- "local" :local
- (throw (IllegalArgumentException. (str "Unknown clojask engine \"" v "\"; use \"onyx\" or \"local\"."))))))
-
(defn- thread-factory
"Daemon threads named prefix-1, prefix-2, ..., so a job that is never
shut down cannot keep the JVM alive."
@@ -94,8 +83,7 @@
(defn- row-worker
"The work on one batch of rows: the columns at index of every row that
passes the filters, evaluated and, if format? is true, formatted. Rows
- that fail a filter are dropped. Same rows as onyx-comps/worker-func-gen
- and worker-func-gen-format."
+ that fail a filter are dropped."
[df exception index format?]
(let [col-info (:col-info df)
operations (.getDesc col-info)
@@ -138,7 +126,7 @@
(throw (stage-error (str "run stage (" stage ")") e)))))
(defn start-compute
- "The local engine's start-onyx: compute df into dist, appending after
+ "Compute df into dist, appending after
the header that DataFrame.compute may have written. Output is always in
input order."
[num-work batch-size df dist exception index melt out]
@@ -152,10 +140,10 @@
"success"))
(defn start-aggre-only
- "The local engine's start-onyx-aggre-only: fold the rows that pass the
+ "Fold the rows that pass the
filters through aggre-func, pairs of a function and a position in
index, then append the result rows to dist. Rows are folded in input
- order. Same output as clojask-aggre."
+ order."
[num-work batch-size df dist exception aggre-func index select out]
(let [work (prepare "aggregate" #(row-worker df exception index false))
memo (volatile! (vec (repeat (count aggre-func) start)))]
@@ -175,9 +163,9 @@
"success"))
(defn start-groupby
- "The local engine's start-onyx-groupby: add every row that passes the
+ "Add every row that passes the
filters to its group, a file under dist when dist is a directory path,
- otherwise the in-memory MGroup dist. Same groups as clojask-groupby."
+ otherwise the in-memory MGroup dist."
[num-work batch-size df dist groupby-keys groupby-index exception & {:keys [format] :or {format false}}]
(let [col-info (:col-info df)
key-index (.getKeyIndex col-info)
@@ -194,7 +182,7 @@
"The work on one group of a group-aggregate, {:file f :d key}: the
group's result rows, the formatted key followed by the aggregates and
reordered by index. f is a group file, or the key in source when the
- groups are in memory. Same rows as aggre-onyx-comps/worker-func-gen."
+ groups are in memory."
[df aggre-funcs index formatters source]
(let [groupby-index (mapv #(nth % 1) (.getGroupbyKeys (:row-info df)))
key-format (u/formatters-by-position (.getFormatter (:col-info df)) groupby-index)
@@ -228,9 +216,8 @@
(.getKeys source)))}))
(defn start-aggre
- "The local engine's start-onyx-aggre: aggregate every group on the pool
- and append the result rows to dist, in the order the groups are listed.
- Same output as aggre-output."
+ "Aggregate every group on the pool and append the result rows to dist,
+ in the order the groups are listed."
[num-work batch-size df source dist exception aggre-func index formatter out]
(let [work (prepare "groupby aggregate" #(group-worker df aggre-func index formatter source))]
(run-stage "groupby aggregate"
@@ -241,10 +228,10 @@
"success"))
(defn start-join
- "The local engine's start-onyx-join: join every row of a that passes the
+ "Join every row of a that passes the
filters to its matches in b, whose groups are in source when they are in
memory, else in the group files under .clojask/join/b/, and append the
- result rows to dist in a's order. Same rows as clojask-join."
+ result rows to dist in a's order."
[num-work batch-size a b source dist exception a-keys b-keys a-roll b-roll join-type limit a-index b-index b-format write-index out]
(let [a-map (.getKeyIndex (:col-info a))
work (prepare "join" #(row-worker a exception (range (count a-map)) false))
@@ -264,8 +251,7 @@
rows of b's group with the same key, or followed by empty cells when b
has none, reordered by write-index. A group is a key of mgroup-a when
the groups are in memory, else a file under .clojask/join/a/; a b group
- file that matched is deleted, so only b's unmatched groups remain. Same
- rows as outer-onyx-comps/worker-func-gen."
+ file that matched is deleted, so only b's unmatched groups remain."
[mgroup-a mgroup-b a-index b-index write-index]
(let [a-new (range (count a-index))
b-new (range (count b-index))
@@ -284,8 +270,7 @@
(defn- outer-b-worker
"The work on one group of b that matched no group of a in an outer join:
- its rows after empty cells for a, reordered by write-index. Same rows as
- outer-onyx-comps/worker-func-gen2."
+ its rows after empty cells for a, reordered by write-index."
[mgroup-b a-index b-index write-index]
(let [b-new (range (count b-index))
a-nil (repeat (count a-index) nil)]
@@ -299,11 +284,10 @@
(map str (rest (file-seq (io/file dir)))))
(defn start-outer
- "The local engine's start-onyx-outer: append to dist every group of a
+ "Append to dist every group of a
joined to b's group with the same key, then every group of b that had
none. The groups are in mgroup-a and mgroup-b when they are in memory,
- else in the group files under .clojask/join/. Same rows as
- outer-output."
+ else in the group files under .clojask/join/."
[num-work batch-size a b mgroup-a mgroup-b dist exception a-index b-index a-format b-format write-index out]
(run-stage "outer join"
#(with-open [wtr (io/writer dist :append true)]
diff --git a/src/main/clojure/clojask/join.clj b/src/main/clojure/clojask/join.clj
index 382516f..434e03f 100644
--- a/src/main/clojure/clojask/join.clj
+++ b/src/main/clojure/clojask/join.clj
@@ -157,8 +157,3 @@
(let [a-row (u/gets-format a-row a-index a-format)]
(write-func writer [(u/gets (concat a-row (repeat count "")) join-index)])))))))
nil))
-
-(defn defn-join
- "Define output-join, the join function of the Onyx join sink."
- [type limit source]
- (def output-join (join-func type limit source)))
diff --git a/src/main/clojure/clojask/join/outer_input.clj b/src/main/clojure/clojask/join/outer_input.clj
deleted file mode 100644
index f3cfc58..0000000
--- a/src/main/clojure/clojask/join/outer_input.clj
+++ /dev/null
@@ -1,96 +0,0 @@
-(ns clojask.join.outer-input
- (:require [onyx.plugin.protocols :as p]
- [clojask.utils]
- [taoensso.timbre :refer [info]]
- [clojure.java.io :as java.io]))
-
-(def mgroup-a nil)
-(def mgroup-b nil)
-
-(defrecord AbsSeqReader [event path rst completed? checkpoint? offset]
- p/Plugin
-
- (start [this event]
- this)
-
- (stop [this event]
- this)
-
- p/Checkpointed
- (checkpoint [this]
- (when checkpoint? @offset))
-
- (recover!
- [this _ checkpoint]
- (vreset! completed? false)
-
- (let [directory (java.io/file path)
- files (if (= mgroup-a nil)
- (rest (file-seq directory))
- (.getKeys mgroup-a))
- data
- (if (= mgroup-a nil)
- (do
- (def tmp (volatile! -1))
- (map (fn [file]
- (vswap! tmp inc)
- {:id @tmp :d (str file)})
- files))
- (do
- (def tmp (volatile! -1))
- (map (fn [file]
- (vswap! tmp inc)
- (if (not= nil mgroup-b) (.delete mgroup-b file))
- {:id @tmp :d (str file)})
- files)))
- ]
- (if (nil? checkpoint)
- (do
- (vreset! rst data)
- (vreset! offset 0))
- (do
- (info "clojask.join.outer-input is recovering state by dropping" checkpoint "elements.")
- (vreset! rst (drop checkpoint data))
- (vreset! offset checkpoint)))))
-
- (checkpointed! [this epoch])
-
- p/BarrierSynchronization
- (synced? [this epoch]
- true)
-
- (completed? [this]
- @completed?)
-
- p/Input
- (poll! [this _ _]
- (if-let [seg (first @rst)]
- (do
- (vswap! rst rest)
- seg)
- (do (vreset! completed? true)
- nil))
- ))
-
-(defn inject-dataframe
- [_mgroup-a _mgroup-b]
- (def mgroup-a _mgroup-a)
- (def mgroup-b _mgroup-b))
-
-(defn input [{:keys [onyx.core/task-map] :as event}]
- (map->AbsSeqReader {:event event
- :path (:buffered-reader/path event)
- :rst (volatile! nil)
- :completed? (volatile! false)
- :checkpoint? (not (false? (:seq/checkpoint? task-map)))
- :offset (volatile! nil)}))
-
-(def reader-calls
- {})
-
-(defn inject-lifecycle-seq
- [_ lifecycle]
- {:seq/seq (:seq/sequential lifecycle)})
-
-(def inject-seq-via-lifecycle
- {:lifecycle/before-task-start inject-lifecycle-seq})
\ No newline at end of file
diff --git a/src/main/clojure/clojask/join/outer_onyx_comps.clj b/src/main/clojure/clojask/join/outer_onyx_comps.clj
deleted file mode 100644
index e0338e3..0000000
--- a/src/main/clojure/clojask/join/outer_onyx_comps.clj
+++ /dev/null
@@ -1,183 +0,0 @@
-(ns clojask.join.outer-onyx-comps
- (:require [clojask.join.outer-input :as input]
- [clojask.join.outer-output :as output]
- [clojask.onyx-comps :as oc]
- [clojure.string :as string]
- [clojask.utils :as u]
- [clojure.java.io :as io]
- [clojask.groupby :refer [read-csv-seq]]))
-
-(def dataframe (atom nil))
-
-(defn worker-func-gen
- [a b mgroup-a mgroup-b exception a-index b-index a-format b-format write-index]
- (let [a-count (count a-index)
- b-count (count b-index)
- b-nil (repeat b-count nil)
- add-nil (fn [row] (concat row b-nil))
- a-index-new (take (count a-index) (iterate inc 0))
- b-index-new (take (count b-index) (iterate inc 0))
- ]
- (if (= nil mgroup-a)
- (defn worker-func
- "refered in preview"
- [seq]
- (let [id (:id seq)
- a-filename (:d seq)
- a-data (read-csv-seq a-filename)
- a-data (map #(u/gets % a-index-new) a-data)
- b-filename (string/replace-first a-filename "/a/" "/b/")]
- (if (.exists (io/file b-filename))
- (do
- (let [b-data (mapv #(u/gets % b-index-new) (read-csv-seq b-filename))]
- (io/delete-file b-filename true)
- {:id id :d (mapv #(u/gets % write-index) (for [a-row a-data b-row b-data] (concat a-row b-row)))}) ;; formatter here
- )
- {:id id :d (mapv #(u/gets % write-index) (map add-nil a-data))})))
- (defn worker-func
- "refered in preview"
- [seq]
- (let [id (:id seq)
- a-filename (:d seq)
- a-data (.getKey mgroup-a a-filename)
- a-data (map #(u/gets % a-index-new) a-data)
- b-filename a-filename
- ]
- (if (.exists mgroup-b b-filename)
- (do
- (let [b-data (mapv #(u/gets % b-index-new) (.getKey mgroup-b b-filename))]
- {:id id :d (mapv #(u/gets % write-index) (for [a-row a-data b-row b-data] (concat a-row b-row)))}) ;; formatter here
- )
- {:id id :d (mapv #(u/gets % write-index) (map add-nil a-data))}))))))
-
-(defn worker-func-gen2
- [a b mgroup-a mgroup-b exception a-index b-index a-format b-format write-index]
- (let [a-count (count a-index)
- b-count (count b-index)
- a-nil (repeat a-count nil)
- add-nil (fn [row] (concat a-nil row))
- b-index-new (take (count b-index) (iterate inc 0))]
- (if (= mgroup-a nil)
- (defn worker-func
- "refered in preview"
- [seq]
- (let [id (:id seq)
- b-filename (:d seq)
- b-data (mapv #(u/gets % b-index-new) (read-csv-seq b-filename))]
- {:id id :d (mapv #(u/gets % write-index) (mapv add-nil b-data))}))
- (defn worker-func
- "refered in preview"
- [seq]
- (let [id (:id seq)
- b-filename (:d seq)
- b-data (mapv #(u/gets % b-index-new) (.getKey mgroup-b b-filename))]
- {:id id :d (mapv #(u/gets % write-index) (mapv add-nil b-data))})))))
-
-(defn catalog-gen
- "Generate the catalog for running Onyx"
- [num-work batch-size]
- ;; initialisation
- (def catalog [])
-
- ;; input
- (def catalog
- (conj catalog
- {:onyx/name :in
- :onyx/plugin :clojask.join.outer-input/input
- :onyx/type :input
- :onyx/medium :seq
- :seq/checkpoint? true
- :onyx/batch-size batch-size
- :onyx/max-peers 1
- :input/doc "Reads segments from a core.async channel"}))
-
- ;; for loop for sample workers
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))
- worker-function (keyword "clojask.join.outer-onyx-comps" "worker-func")]
- (def catalog
- (conj catalog
- {:onyx/name worker-name
- :onyx/fn worker-function
- :onyx/type :function
- :onyx/batch-size batch-size
- :worker/doc "This is a worker node"}
- ))))
-
- ;; output
- (def catalog
- (conj catalog
- {:onyx/name :output
- :onyx/plugin :clojask.join.outer-output/output
- :onyx/type :output
- :onyx/medium :core.async ;; this is maked up
- :onyx/max-peers 1
- :onyx/batch-size batch-size
- :output/doc "Writes segments to the file"}))
-
- )
-
-(defn inject-in-reader [event lifecycle]
- (let []
- {:buffered-reader/path (:buffered-reader/path lifecycle)
- }))
-
-(def in-calls
- {:lifecycle/before-task-start inject-in-reader})
-
-(defn lifecycle-gen
- [source dist]
- (def lifecycles
- [{:lifecycle/task :in
- :buffered-reader/path source
- :lifecycle/calls ::in-calls}
- {:lifecycle/task :in
- :lifecycle/calls :clojask.join.outer-input/reader-calls}
- {:lifecycle/task :output
- :buffered-wtr/filename dist
- :lifecycle/calls :clojask.join.outer-output/writer-calls}]))
-
-(defn flow-cond-gen
- "Generate the flow conditions for running Onyx"
- [num-work]
- (def flow-conditions []) ;; initialisation
-
- ;; for loop for sample workers
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))
- predicate-function (keyword "clojask.join.outer-onyx-comps" (str "rem" (- x 1) "?"))]
- (intern 'clojask.join.outer-onyx-comps (symbol (str "rem" (- x 1) "?")) (fn [event old-segment new-segment all-new-segment]
- (= (mod (:id new-segment) num-work) (- x 1))))
- (def flow-conditions
- (conj flow-conditions
- {:flow/from :in
- :flow/to [worker-name]
- :flow/predicate predicate-function
- :worker/doc "This is a flow condition"}
- ))))
-
- )
-
-(defn start-onyx-outer
- "start the onyx cluster with the specification inside dataframe"
- [num-work batch-size a b mgroup-a mgroup-b dist exception a-index b-index a-format b-format write-index output]
- (oc/with-onyx-env "outer join" num-work
- (fn []
- ;; step 1: every row of a, with its matches in b
- (oc/run-job "outer join"
- (fn []
- (worker-func-gen a b mgroup-a mgroup-b exception a-index b-index a-format b-format write-index)
- (catalog-gen num-work batch-size)
- (lifecycle-gen "./.clojask/join/a" dist)
- (flow-cond-gen num-work)
- (input/inject-dataframe mgroup-a mgroup-b)
- (output/inject-write-func output)
- {:catalog catalog :lifecycles lifecycles :flow-conditions flow-conditions}))
- ;; step 2: the rows of b that had no match, on the same environment
- (oc/run-job "outer join 2"
- (fn []
- (if (not= mgroup-b nil) (.final mgroup-b))
- (worker-func-gen2 a b mgroup-a mgroup-b exception a-index b-index a-format b-format write-index)
- (lifecycle-gen "./.clojask/join/b" dist)
- (input/inject-dataframe mgroup-b nil)
- {:catalog catalog :lifecycles lifecycles :flow-conditions flow-conditions})))))
\ No newline at end of file
diff --git a/src/main/clojure/clojask/join/outer_output.clj b/src/main/clojure/clojask/join/outer_output.clj
deleted file mode 100644
index d5cb833..0000000
--- a/src/main/clojure/clojask/join/outer_output.clj
+++ /dev/null
@@ -1,84 +0,0 @@
-(ns clojask.join.outer-output
- (:require [onyx.plugin.protocols :as p]
- [clojure.java.io :as io]))
-
-(def write-func (atom nil))
-
-(defn inject-write-func
- [func]
- (reset! write-func func))
-
-(defn- inject-into-eventmap
- [event lifecycle]
- (let [wtr (io/writer (:buffered-wtr/filename lifecycle) :append true)]
- {:clojask/wtr wtr}))
-
-(defn- close-writer [event lifecycle]
- (.close (:clojask/wtr event)))
-
-;; Map of lifecycle calls that are required to use this plugin.
-;; Users will generally always have to include these in their lifecycle calls
-;; when submitting the job.
-(def writer-calls
- {:lifecycle/before-task-start inject-into-eventmap
- :lifecycle/after-task-stop close-writer})
-
-(defrecord ClojaskOutput [write-func]
- p/Plugin
- (start [this event]
- ;; Initialize the plugin, generally by assoc'ing any initial state.
- this)
-
- (stop [this event]
- ;; Nothing is required here. However, most plugins have resources
- ;; (e.g. a connection) to clean up.
- ;; Mind that such cleanup is also achievable with lifecycles.
- this)
-
- p/Checkpointed
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpoint [this])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (recover! [this replica-version checkpoint])
-
- ;; Nothing is required here. This is normally useful for checkpointing in
- ;; input plugins.
- (checkpointed! [this epoch])
-
- p/BarrierSynchronization
- (synced? [this epoch]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished.
- true)
-
- (completed? [this]
- ;; Nothing is required here. This is commonly used to check whether all
- ;; async writes have finished (just like synced).
- true)
-
- p/Output
- (prepare-batch [this event replica messenger]
- ;; Nothing is required here. This is useful for some initial preparation,
- ;; before write-batch is called repeatedly.
- true)
-
- (write-batch [this {:keys [onyx.core/write-batch clojask/wtr]} replica messenger]
- (doseq [msg write-batch]
- (do
- (if (not= (:d msg) nil)
- (do
- (write-func wtr (:d msg))
- ))))
- (.flush wtr)
- true))
-
-;; Builder function for your output plugin.
-;; Instantiates a record.
-;; It is highly recommended you inject and pre-calculate frequently used data
-;; from your task-map here, in order to improve the performance of your plugin
-;; Extending the function below is likely good for most use cases.
-(defn output [pipeline-data]
- (->ClojaskOutput (deref write-func)))
\ No newline at end of file
diff --git a/src/main/clojure/clojask/onyx_comps.clj b/src/main/clojure/clojask/onyx_comps.clj
deleted file mode 100644
index 78963e9..0000000
--- a/src/main/clojure/clojask/onyx_comps.clj
+++ /dev/null
@@ -1,592 +0,0 @@
-(ns clojask.onyx-comps
- (:require [clojask.clojask-aggre :as aggre]
- [clojask.clojask-groupby :as groupby]
- [clojask.clojask-input :as input]
- [clojask.clojask-join :as join]
- [clojask.clojask-output :as output]
- [clojask.join :refer [defn-join]]
- [clojask.utils :as u]
- [onyx.api]
- [onyx.test-helper :refer [feedback-exception!]]
- [taoensso.timbre.appenders.3rd-party.rotor :as rotor])
- (:import [com.clojask.exception ExecutionException]))
-
-;; sample workflow
-;;
-;; [[:in :sample-worker1]
-;; [:in :sample-worker2]
-;; [:sample-worker1 :output]
-;; [:sample-worker2 :output]]
-
-(def id (java.util.UUID/randomUUID))
-
-(defn workflow-gen
- "Generate workflow for running Onyx"
- [num-work]
- (def workflow []) ;; initialisation
-
- ;; for loop for input edges
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))]
- (def workflow (conj workflow [:in worker-name]
- ))))
-
- ;; for loop for output edges
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))]
- (def workflow (conj workflow [worker-name :output]
- ))))
-
- )
-
-(def dataframe (atom nil))
-
-(defn worker-func-gen
- [df exception index]
- (reset! dataframe df)
- (let [operations (.getDesc (:col-info (deref dataframe)))
- types (.getType (:col-info (deref dataframe)))
- formats (.getFormatter (:col-info (deref dataframe)))
- filters (.getFilters (:row-info df))
- indices index]
- (if exception
- (defn worker-func
- [seg]
- (let [id (:id seg)
- data (:d seg)] ;; -1 is very important here!
- {:id id :d (for [row data]
- (if (u/filter-check filters types row)
- (mapv (fn [_] (u/eval-res row types formats operations _)) indices)
- nil))}
- ))
- (defn worker-func
- [seg]
- (let [id (:id seg)
- data (:d seg)]
- {:id id :d (for [row data]
- (if (u/filter-check filters types row)
- (mapv (fn [_] (u/eval-res-ne row types formats operations _)) indices)
- nil))}
- ))))
- )
-
-(defn worker-func-gen-format
- [df exception index]
- (reset! dataframe df)
- (let [operations (.getDesc (:col-info (deref dataframe)))
- types (.getType (:col-info (deref dataframe)))
- formats (.getFormatter (:col-info (deref dataframe)))
- filters (.getFilters (:row-info df))
- indices index]
- (if exception
- (defn worker-func
- [seg]
- (let [id (:id seg)
- data (:d seg)] ;; -1 is very important here!
- {:id id :d (for [row data]
- (if (u/filter-check filters types row)
- (mapv (fn [_] ((or (get formats _) str) (u/eval-res row types formats operations _))) indices)
- nil))}
- ))
- (defn worker-func
- [seg]
- (let [id (:id seg)
- data (:d seg)]
- {:id id :d (for [row data]
- (if (u/filter-check filters types row)
- (mapv (fn [_] ((or (get formats _) str) (u/eval-res-ne row types formats operations _))) indices)
- nil))}
- )))))
-
-(defn catalog-gen
- "Generate the catalog for running Onyx"
- [num-work batch-size]
- ;; initialisation
- (def catalog [])
-
- ;; input
- (def catalog
- (conj catalog
- {:onyx/name :in
- :onyx/plugin :clojask.clojask-input/input
- :onyx/type :input
- :onyx/medium :seq
- :seq/checkpoint? true
- :onyx/batch-size batch-size
- :onyx/max-peers 1
- :input/doc "Reads segments from a core.async channel"}))
-
- ;; for loop for sample workers
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))
- worker-function (keyword "clojask.onyx-comps" "worker-func")]
- (def catalog
- (conj catalog
- {:onyx/name worker-name
- :onyx/fn worker-function
- :onyx/type :function
- :onyx/batch-size batch-size
- :worker/doc "This is a worker node"}
- ))))
-
- ;; output
- (def catalog
- (conj catalog
- {:onyx/name :output
- :onyx/plugin :clojask.clojask-output/output
- :onyx/type :output
- :onyx/medium :core.async ;; this is maked up
- :onyx/max-peers 1
- :onyx/batch-size batch-size
- :output/doc "Writes segments to the file"}))
-
- )
-
-(defn catalog-aggre-gen
- "Generate the catalog for running Onyx"
- [num-work batch-size]
- ;; initialisation
- (def catalog [])
-
- ;; input
- (def catalog
- (conj catalog
- {:onyx/name :in
- :onyx/plugin :clojask.clojask-input/input
- :onyx/type :input
- :onyx/medium :seq
- :seq/checkpoint? true
- :onyx/batch-size batch-size
- :onyx/max-peers 1
- :input/doc "Reads segments from a core.async channel"}))
-
- ;; for loop for sample workers
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))
- worker-function (keyword "clojask.onyx-comps" "worker-func")]
- (def catalog
- (conj catalog
- {:onyx/name worker-name
- :onyx/fn worker-function
- :onyx/type :function
- :onyx/batch-size batch-size
- :worker/doc "This is a worker node"}))))
-
- ;; output
- (def catalog
- (conj catalog
- {:onyx/name :output
- :onyx/plugin :clojask.clojask-aggre/output
- :onyx/type :output
- :onyx/medium :core.async ;; this is maked up
- :onyx/max-peers 1
- :onyx/batch-size batch-size
- :output/doc "Writes segments to the file"}))
-
- )
-
-(defn catalog-groupby-gen
- "Generate the catalog for running Onyx"
- [num-work batch-size]
- ;; initialisation
- (def catalog [])
-
- ;; input
- (def catalog
- (conj catalog
- {:onyx/name :in
- :onyx/plugin :clojask.clojask-input/input
- :onyx/type :input
- :onyx/medium :seq
- :seq/checkpoint? true
- :onyx/batch-size batch-size
- :onyx/max-peers 1
- :input/doc "Reads segments from a core.async channel"}))
-
- ;; for loop for sample workers
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))
- worker-function (keyword "clojask.onyx-comps" "worker-func")]
- (def catalog
- (conj catalog
- {:onyx/name worker-name
- :onyx/fn worker-function
- :onyx/type :function
- :onyx/batch-size batch-size
- :worker/doc "This is a worker node"}))))
-
- ;; output
- (def catalog
- (conj catalog
- {:onyx/name :output
- :onyx/plugin :clojask.clojask-groupby/groupby
- :onyx/type :output
- :onyx/medium :core.async ;; this is maked up
- :onyx/max-peers 1
- :onyx/batch-size batch-size
- :output/doc "Writes segments to the file"}))
-
- )
-
-(defn catalog-join-gen
- "Generate the catalog for running Onyx"
- [num-work batch-size]
- ;; initialisation
- (def catalog [])
-
- ;; input
- (def catalog
- (conj catalog
- {:onyx/name :in
- :onyx/plugin :clojask.clojask-input/input
- :onyx/type :input
- :onyx/medium :seq
- :seq/checkpoint? true
- :onyx/batch-size batch-size
- :onyx/max-peers 1
- :input/doc "Reads segments from a core.async channel"}))
-
- ;; for loop for sample workers
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))
- worker-function (keyword "clojask.onyx-comps" "worker-func")]
- (def catalog
- (conj catalog
- {:onyx/name worker-name
- :onyx/fn worker-function
- :onyx/type :function
- :onyx/batch-size batch-size
- :worker/doc "This is a worker node"}))))
-
- ;; output
- (def catalog
- (conj catalog
- {:onyx/name :output
- :onyx/plugin :clojask.clojask-join/join
- :onyx/type :output
- :onyx/medium :core.async ;; this is maked up
- :onyx/max-peers 1
- :onyx/batch-size batch-size
- :output/doc "Writes segments to the file"}))
-
- )
-
-(defn inject-in-reader [event lifecycle]
- {})
-
-(defn close-reader [event lifecycle]
- (if (not= (:seq/rdr event) nil)
- (.close (:seq/rdr event))))
-
-(def in-calls
- {:lifecycle/before-task-start inject-in-reader
- :lifecycle/after-task-stop close-reader})
-
-(defn lifecycle-gen
- [source dist order select]
- (def lifecycles
- [{:lifecycle/task :in
- :buffered-reader/filename nil
- :lifecycle/calls ::in-calls}
- {:lifecycle/task :in
- :lifecycle/calls :clojask.clojask-input/reader-calls}
- {:lifecycle/task :output
- :buffered-wtr/filename dist
- :order order
- :indices select
- :lifecycle/calls :clojask.clojask-output/writer-calls}]))
-
-(defn lifecycle-aggre-gen
- [source dist]
- (def lifecycles
- [{:lifecycle/task :in
- :buffered-reader/filename (if (fn? source) nil source)
- :lifecycle/calls ::in-calls}
- {:lifecycle/task :in
- :lifecycle/calls :clojask.clojask-input/reader-calls}
- {:lifecycle/task :output
- :buffered-wtr/filename dist
- :lifecycle/calls :clojask.clojask-aggre/writer-calls}]))
-
-(defn lifecycle-groupby-gen
- [source dist keys key-index]
- (def lifecycles
- [{:lifecycle/task :in
- :buffered-reader/filename (if (fn? source) nil source)
- :lifecycle/calls ::in-calls}
- {:lifecycle/task :in
- :lifecycle/calls :clojask.clojask-input/reader-calls}
- {:lifecycle/task :output
- :buffered-wtr/filename dist
- :clojask/key-index key-index
- :lifecycle/calls :clojask.clojask-groupby/writer-aggre-calls}]))
-
-(defn lifecycle-join-gen
- [source dist a b a-keys b-keys a-roll b-roll join-type]
- (def lifecycles
- [{:lifecycle/task :in
- :buffered-reader/filename (if (fn? source) nil source)
- :lifecycle/calls ::in-calls}
- {:lifecycle/task :in
- :lifecycle/calls :clojask.clojask-input/reader-calls}
- {:lifecycle/task :output
- :buffered-wtr/filename dist
- :clojask/a-roll a-roll
- :clojask/b-roll b-roll
- :clojask/a-map (.getKeyIndex (.col-info a))
- :clojask/b-map (.getKeyIndex (.col-info b))
- :clojask/join-type join-type
- :lifecycle/calls :clojask.clojask-join/writer-join-calls}]))
-
-(defn flow-cond-gen
- "Generate the flow conditions for running Onyx"
- [num-work]
- (def flow-conditions []) ;; initialisation
- (def predicate-funcs [])
- ;; for loop for sample workers
- (doseq [x (range 1 (+ num-work 1))]
- (let [worker-name (keyword (str "sample-worker" x))
- predicate-function (keyword "clojask.onyx-comps" (str "rem" (- x 1) "?"))
- ]
- (intern 'clojask.onyx-comps (symbol (str "rem" (- x 1) "?")) (fn [event old-segment new-segment all-new-segment]
- (= (mod (:id new-segment) num-work) (- x 1))))
- (def flow-conditions
- (conj flow-conditions
- {:flow/from :in
- :flow/to [worker-name]
- :flow/predicate predicate-function
- :worker/doc "This is a flow condition"}))))
-
- )
-
-;; Components started by config-env. Kept as top-level vars so shutdown can
-;; tear down whatever a partial startup managed to create.
-(def env nil)
-(def peer-group nil)
-(def v-peers nil)
-
-(defn- setting
- "The JVM system property prop, else the environment variable env, else
- default."
- [prop env default]
- (or (System/getProperty prop) (System/getenv env) default))
-
-(defn- port-setting
- [prop env default]
- (let [v (setting prop env default)]
- (if (string? v) (Long/parseLong v) v)))
-
-(def log-path ".clojask/clojask.log")
-
-(defn- log-config
- "timbre configuration handed to Onyx: warnings and errors go to the log
- file, rotated at 10 MB with one backup, and errors also to stdout.
- Onyx's own default logs at :info, which adds a few hundred KB per
- compute and keeps up to 300 MB of files. Only the appenders get levels:
- the global level belongs to clojask.dataframe, so enable-debug stays in
- effect across computes."
- []
- {:appenders {:println {:min-level :error :enabled? true}
- :rotor (assoc (rotor/rotor-appender {:path log-path
- :max-size (* 10 1024 1024)
- :backlog 1})
- :min-level :warn)}})
-
-(defn config-env
- "Start the embedded ZooKeeper, the peer group and the peers for the
- current workflow. The ZooKeeper port, the Aeron port and the Aeron
- media-driver directory come from the JVM system properties
- clojask.zookeeper.port, clojask.aeron.port and clojask.aeron.dir, or
- the environment variables CLOJASK_ZOOKEEPER_PORT, CLOJASK_AERON_PORT
- and CLOJASK_AERON_DIR, so two clojask processes can share a machine."
- []
- ;; Clear the previous run first, so a failure below leaves only the
- ;; components that were actually started in this run.
- (def env nil)
- (def peer-group nil)
- (def v-peers nil)
- (let [zk-port (port-setting "clojask.zookeeper.port" "CLOJASK_ZOOKEEPER_PORT" 2188)
- aeron-port (port-setting "clojask.aeron.port" "CLOJASK_AERON_PORT" 40200)
- aeron-dir (setting "clojask.aeron.dir" "CLOJASK_AERON_DIR" nil)
- zk-address (str "127.0.0.1:" zk-port)
- log (log-config)]
- (def env-config
- {:zookeeper/address zk-address
- :zookeeper/server? true
- :zookeeper.server/port zk-port
- :onyx/tenancy-id id
- :onyx.log/file log-path
- :onyx.log/config log})
- (def peer-config
- (cond-> {:zookeeper/address zk-address
- :onyx/tenancy-id id
- :onyx.peer/job-scheduler :onyx.job-scheduler/balanced
- :onyx.messaging/impl :aeron
- :onyx.messaging/peer-port aeron-port
- :onyx.messaging/bind-addr "localhost"
- :onyx.log/file log-path
- :onyx.log/config log}
- aeron-dir (assoc :onyx.messaging.aeron/media-driver-dir aeron-dir))))
-
- (def env (onyx.api/start-env env-config))
-
- (def peer-group (onyx.api/start-peer-group peer-config))
-
- (def n-peers (count (set (mapcat identity workflow))))
-
- (def v-peers (onyx.api/start-peers n-peers peer-group)))
-
-(defn shutdown
- "Tear down whatever config-env managed to start, in reverse order. Every
- step is attempted even if an earlier one fails, so a startup that dies
- after ZooKeeper is up (for example in the peer group) still releases the
- ZooKeeper port instead of poisoning every later compute in this JVM. The
- first error, if any, is rethrown once all steps have run."
- []
- (let [errors (volatile! [])
- attempt (fn [f] (try (f) (catch Exception e (vswap! errors conj e))))]
- (doseq [v-peer v-peers]
- (attempt #(onyx.api/shutdown-peer v-peer)))
- (when peer-group
- (attempt #(onyx.api/shutdown-peer-group peer-group)))
- (when env
- (attempt #(onyx.api/shutdown-env env)))
- (def v-peers nil)
- (def peer-group nil)
- (def env nil)
- (when-let [e (first @errors)]
- (throw e))))
-
-(defn- stage-error
- [stage e]
- (ExecutionException.
- (format "[%s] Refer to .clojask/clojask.log for detailed information. (original error: %s)"
- stage (.getMessage e))
- e))
-
-(defn submit-and-wait
- "Submit one job to the running environment and block until it has
- finished, rethrowing the job's own exception if it failed. The job map
- supplies :catalog, :lifecycles and :flow-conditions; the workflow is the
- one config-env sized the peers for."
- [job]
- (let [submission (onyx.api/submit-job peer-config
- (assoc job
- :workflow workflow
- :task-scheduler :onyx.task-scheduler/balanced))
- job-id (:job-id submission)]
- (assert job-id "Job was not successfully submitted")
- (feedback-exception! peer-config job-id)))
-
-(defn run-job
- "Run one job inside with-onyx-env. prepare! does the per-job setup and
- returns the job map for submit-and-wait. Failures are tagged with the
- stage they came from; with-onyx-env shuts the environment down."
- [stage prepare!]
- (let [job (try
- (prepare!)
- (catch Throwable e
- (throw (stage-error (str "preparing stage (" stage ")") e))))]
- (try
- (submit-and-wait job)
- (catch Throwable e
- (throw (stage-error (str "submit-to-onyx stage (" stage ")") e))))))
-
-(defn with-onyx-env
- "Start the Onyx environment (embedded ZooKeeper, peer group and one peer
- per task) for num-work workers, call f, and shut everything down again
- whether or not f threw. This is the only place that starts or stops the
- environment, so every driver gets the same partial-startup handling."
- [stage num-work f]
- (try
- (workflow-gen num-work)
- (config-env)
- (catch Exception e
- (try (shutdown) (catch Exception _))
- (throw (stage-error (str "preparing stage (" stage ")") e))))
- (try
- (f)
- ;; Throwable: an AssertionError from a worker must still shut Onyx down
- (catch Throwable e
- (try (shutdown) (catch Exception _))
- (throw (if (instance? ExecutionException e)
- e
- (stage-error (str "run stage (" stage ")") e)))))
- (try
- (shutdown)
- (catch Exception e
- (throw (stage-error (str "terminate-node stage (" stage ")") e))))
- "success")
-
-(defn- job-spec
- "The job map built by the *-gen functions of this namespace."
- []
- {:catalog catalog :lifecycles lifecycles :flow-conditions flow-conditions})
-
-(defn start-onyx
- "start the onyx cluster with the specification inside dataframe"
- [num-work batch-size dataframe dist exception order index melt out]
- (with-onyx-env "compute" num-work
- (fn []
- (run-job "compute"
- (fn []
- (worker-func-gen-format dataframe exception index)
- (catalog-gen num-work batch-size)
- (lifecycle-gen (.getFunc dataframe) dist order index)
- (flow-cond-gen num-work)
- (input/inject-dataframe dataframe)
- (output/inject-dataframe dataframe out)
- (output/inject-melt melt)
- (job-spec))))))
-
-(defn start-onyx-aggre-only
- "start the onyx cluster with the specification inside dataframe"
- [num-work batch-size dataframe dist exception aggre-func index select out]
- (with-onyx-env "aggregate" num-work
- (fn []
- (run-job "aggregate"
- (fn []
- (worker-func-gen dataframe exception index)
- (catalog-aggre-gen num-work batch-size)
- (lifecycle-aggre-gen (.getFunc dataframe) dist)
- (flow-cond-gen num-work)
- (input/inject-dataframe dataframe)
- (aggre/inject-dataframe dataframe aggre-func select out)
- (job-spec))))))
-
-(defn start-onyx-groupby
- "start the onyx cluster with the specification inside dataframe\n
- @format: if format the value before writing to file. For procedures that will need to compare / use the actual
- value of each element, should be set to false, such as aggregate, rolling join. For others, should be set to
- false to avoid repeated formatting
- "
- [num-work batch-size dataframe dist groupby-keys groupby-index exception & {:keys [format] :or {format false}}]
- (with-onyx-env "groupby" num-work
- (fn []
- (run-job "groupby"
- (fn []
- (worker-func-gen dataframe exception (vec (take (count (.getKeyIndex (.col-info dataframe))) (iterate inc 0))))
- (catalog-groupby-gen num-work batch-size)
- ;; use of dist from here is deprecated
- (lifecycle-groupby-gen (.getFunc dataframe) (if (string? dist) dist nil) groupby-keys (.getKeyIndex (.col-info dataframe)))
- (flow-cond-gen num-work)
- (input/inject-dataframe dataframe)
- (groupby/inject-dataframe dataframe groupby-keys groupby-index dist format)
- (job-spec))))))
-
-(defn start-onyx-join
- "start the onyx cluster with the specification inside dataframe"
- [num-work batch-size dataframe b source dist exception a-keys b-keys a-roll b-roll join-type limit a-index b-index b-format write-index out]
- ;; dataframe means a
- (with-onyx-env "join" num-work
- (fn []
- (run-job "join"
- (fn []
- (worker-func-gen dataframe exception (take (count (.getKeyIndex (:col-info dataframe))) (iterate inc 0)))
- (catalog-join-gen num-work batch-size)
- (lifecycle-join-gen (.getFunc dataframe) dist dataframe b a-keys b-keys a-roll b-roll join-type)
- (flow-cond-gen num-work)
- (input/inject-dataframe dataframe)
- (join/inject-dataframe dataframe b a-keys b-keys a-index b-index write-index b-format out)
- (defn-join join-type (or limit (fn [a b] true)) source)
- (job-spec))))))
diff --git a/test/clojask/api_test.clj b/test/clojask/api_test.clj
index 7b73c01..9cda765 100644
--- a/test/clojask/api_test.clj
+++ b/test/clojask/api_test.clj
@@ -1,6 +1,6 @@
(ns api-test
- "Argument validation in the public API. Nothing here starts Onyx, so
- these run in well under a second."
+ "Argument validation in the public API. Every compute here is rejected
+ before it starts, so these run in well under a second."
(:require [clojure.test :refer :all]
[clojure.java.io :as io]
[clojure.string :as str]
diff --git a/test/clojask/core_test.clj b/test/clojask/core_test.clj
index 909b98b..394e9f9 100644
--- a/test/clojask/core_test.clj
+++ b/test/clojask/core_test.clj
@@ -15,8 +15,9 @@
(defn get-diff
"Compare an output file with the expected one. The header line must match
- exactly; data rows are compared sorted (Onyx writes them in any order)
- unless order is false."
+ exactly; data rows are compared sorted (group-by and outer-join output
+ comes in group order, which depends on the file system) unless order
+ is false."
[a b & [order]]
(try
(let [order (if (nil? order) true order)
diff --git a/test/clojask/correct_outputs/engine.edn b/test/clojask/correct_outputs/engine.edn
new file mode 100644
index 0000000..7335385
--- /dev/null
+++ b/test/clojask/correct_outputs/engine.edn
@@ -0,0 +1,301 @@
+;; engine-test's expected outputs, recorded from Onyx before 2.1.0 removed it
+;; (the thread-pool engine wrote the same): per case the header, then the rows sorted.
+{"aggregate-only" ["max,min,n" "800.0,300.0,4"],
+ "group-aggregate, in-memory false"
+ ["Department,max,mean"
+ "11,50000.0,16966.666666666668"
+ "12,1000.0,950.0"
+ "13,800.0,800.0"
+ "21,700.0,700.0"],
+ "group-aggregate, in-memory true"
+ ["Department,max,mean"
+ "11,50000.0,16966.666666666668"
+ "12,1000.0,950.0"
+ "13,800.0,800.0"
+ "21,700.0,700.0"],
+ "group-aggregate-multi-row-result"
+ ["low,Department"
+ "1000.0,12"
+ "300.0,11"
+ "50000.0,11"
+ "600.0,11"
+ "700.0,21"
+ "800.0,13"
+ "900.0,12"],
+ "group-by-only" ["Department" "11" "12" "13" "21"],
+ "join inner swapped, in-memory false"
+ ["1_Employee,1_EmployeeName,1_DayOff,1_UpdateDate,2_Employee,2_EmployeeName,2_Department,2_Salary,2_UpdateDate"
+ "1,Alice,d20,2020/12/10,1,Alice,11,-300.0!,2020/12/12"
+ "2,Bob,d15,2020/12/05,2,Bob,11,-600.0!,2020/12/01"
+ "3,Carla,d5,2020/12/03,3,Carla,12,-900.0!,2020/12/03"],
+ "join inner swapped, in-memory true"
+ ["1_Employee,1_EmployeeName,1_DayOff,1_UpdateDate,2_Employee,2_EmployeeName,2_Department,2_Salary,2_UpdateDate"
+ "1,Alice,d20,2020/12/10,1,Alice,11,-300.0!,2020/12/12"
+ "2,Bob,d15,2020/12/05,2,Bob,11,-600.0!,2020/12/01"
+ "3,Carla,d5,2020/12/03,3,Carla,12,-900.0!,2020/12/03"],
+ "join inner, in-memory false"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,1,Alice,d20,2020/12/10"
+ "2,Bob,11,-600.0!,2020/12/01,2,Bob,d15,2020/12/05"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"],
+ "join inner, in-memory true"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,1,Alice,d20,2020/12/10"
+ "2,Bob,11,-600.0!,2020/12/01,2,Bob,d15,2020/12/05"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"],
+ "join left, in-memory false"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,1,Alice,d20,2020/12/10"
+ "2,Bob,11,-600.0!,2020/12/01,2,Bob,d15,2020/12/05"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join left, in-memory true"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,1,Alice,d20,2020/12/10"
+ "2,Bob,11,-600.0!,2020/12/01,2,Bob,d15,2020/12/05"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join outer swapped, in-memory false"
+ ["1_Employee,1_EmployeeName,1_DayOff,1_UpdateDate,2_Employee,2_EmployeeName,2_Department,2_Salary,2_UpdateDate"
+ ",,,,4,Daniel,12,-1000.0!,2020/12/05"
+ ",,,,5,Evelyn,13,-800.0!,2020/12/03"
+ ",,,,6,Ferdinand,21,-700.0!,2020/12/05"
+ "1,Alice,d20,2020/12/10,1,Alice,11,-300.0!,2020/12/12"
+ "2,Bob,d15,2020/12/05,2,Bob,11,-600.0!,2020/12/01"
+ "3,Carla,d5,2020/12/03,3,Carla,12,-900.0!,2020/12/03"
+ "7,Angel,d30,2020/12/11,,,,,"
+ "8,Jack,d4,2019/03/21,,,,,"],
+ "join outer swapped, in-memory true"
+ ["1_Employee,1_EmployeeName,1_DayOff,1_UpdateDate,2_Employee,2_EmployeeName,2_Department,2_Salary,2_UpdateDate"
+ ",,,,4,Daniel,12,-1000.0!,2020/12/05"
+ ",,,,5,Evelyn,13,-800.0!,2020/12/03"
+ ",,,,6,Ferdinand,21,-700.0!,2020/12/05"
+ "1,Alice,d20,2020/12/10,1,Alice,11,-300.0!,2020/12/12"
+ "2,Bob,d15,2020/12/05,2,Bob,11,-600.0!,2020/12/01"
+ "3,Carla,d5,2020/12/03,3,Carla,12,-900.0!,2020/12/03"
+ "7,Angel,d30,2020/12/11,,,,,"
+ "8,Jack,d4,2019/03/21,,,,,"],
+ "join outer, in-memory false"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ ",,,,,7,Angel,d30,2020/12/11"
+ ",,,,,8,Jack,d4,2019/03/21"
+ "1,Alice,11,-300.0!,2020/12/12,1,Alice,d20,2020/12/10"
+ "2,Bob,11,-600.0!,2020/12/01,2,Bob,d15,2020/12/05"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join outer, in-memory true"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ ",,,,,7,Angel,d30,2020/12/11"
+ ",,,,,8,Jack,d4,2019/03/21"
+ "1,Alice,11,-300.0!,2020/12/12,1,Alice,d20,2020/12/10"
+ "2,Bob,11,-600.0!,2020/12/01,2,Bob,d15,2020/12/05"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join right, in-memory false"
+ ["2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate,1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate"
+ "1,Alice,d20,2020/12/10,1,Alice,11,-300.0!,2020/12/12"
+ "2,Bob,d15,2020/12/05,2,Bob,11,-600.0!,2020/12/01"
+ "3,Carla,d5,2020/12/03,3,Carla,12,-900.0!,2020/12/03"
+ "7,Angel,d30,2020/12/11,,,,,"
+ "8,Jack,d4,2019/03/21,,,,,"],
+ "join right, in-memory true"
+ ["2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate,1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate"
+ "1,Alice,d20,2020/12/10,1,Alice,11,-300.0!,2020/12/12"
+ "2,Bob,d15,2020/12/05,2,Bob,11,-600.0!,2020/12/01"
+ "3,Carla,d5,2020/12/03,3,Carla,12,-900.0!,2020/12/03"
+ "7,Angel,d30,2020/12/11,,,,,"
+ "8,Jack,d4,2019/03/21,,,,,"],
+ "join rolling backward with limit, in-memory false"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,,,,"
+ "2,Bob,11,-600.0!,2020/12/01,,,,"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join rolling backward with limit, in-memory true"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,,,,"
+ "2,Bob,11,-600.0!,2020/12/01,,,,"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join rolling backward, in-memory false"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,,,,"
+ "2,Bob,11,-600.0!,2020/12/01,2,Bob,d15,2020/12/05"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join rolling backward, in-memory true"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,,,,"
+ "2,Bob,11,-600.0!,2020/12/01,2,Bob,d15,2020/12/05"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join rolling forward, in-memory false"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,1,Alice,d20,2020/12/10"
+ "2,Bob,11,-600.0!,2020/12/01,,,,"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join rolling forward, in-memory true"
+ ["1_Employee,1_EmployeeName,1_Department,1_Salary,1_UpdateDate,2_Employee,2_EmployeeName,2_DayOff,2_UpdateDate"
+ "1,Alice,11,-300.0!,2020/12/12,1,Alice,d20,2020/12/10"
+ "2,Bob,11,-600.0!,2020/12/01,,,,"
+ "3,Carla,12,-900.0!,2020/12/03,3,Carla,d5,2020/12/03"
+ "4,Daniel,12,-1000.0!,2020/12/05,,,,"
+ "5,Evelyn,13,-800.0!,2020/12/03,,,,"
+ "6,Ferdinand,21,-700.0!,2020/12/05,,,,"],
+ "join-select inner swapped, in-memory false"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ "2020/12/01,2020/12/05,2"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/12,2020/12/10,1"],
+ "join-select inner swapped, in-memory true"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ "2020/12/01,2020/12/05,2"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/12,2020/12/10,1"],
+ "join-select inner, in-memory false"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"
+ "2020/12/10,2020/12/12,1"],
+ "join-select inner, in-memory true"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"
+ "2020/12/10,2020/12/12,1"],
+ "join-select left, in-memory false"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"
+ "2020/12/10,2020/12/12,1"],
+ "join-select left, in-memory true"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"
+ "2020/12/10,2020/12/12,1"],
+ "join-select outer swapped, in-memory false"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2019/03/21,8"
+ ",2020/12/11,7"
+ "2020/12/01,2020/12/05,2"
+ "2020/12/03,,"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,,"
+ "2020/12/05,,"
+ "2020/12/12,2020/12/10,1"],
+ "join-select outer swapped, in-memory true"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2019/03/21,8"
+ ",2020/12/11,7"
+ "2020/12/01,2020/12/05,2"
+ "2020/12/03,,"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,,"
+ "2020/12/05,,"
+ "2020/12/12,2020/12/10,1"],
+ "join-select outer, in-memory false"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ "2019/03/21,,"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"
+ "2020/12/10,2020/12/12,1"
+ "2020/12/11,,"],
+ "join-select outer, in-memory true"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ "2019/03/21,,"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"
+ "2020/12/10,2020/12/12,1"
+ "2020/12/11,,"],
+ "join-select right, in-memory false"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ "2019/03/21,,"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"
+ "2020/12/10,2020/12/12,1"
+ "2020/12/11,,"],
+ "join-select right, in-memory true"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ "2019/03/21,,"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"
+ "2020/12/10,2020/12/12,1"
+ "2020/12/11,,"],
+ "join-select rolling backward with limit, in-memory false"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/01,2"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ ",2020/12/12,1"
+ "2020/12/03,2020/12/03,3"],
+ "join-select rolling backward with limit, in-memory true"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/01,2"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ ",2020/12/12,1"
+ "2020/12/03,2020/12/03,3"],
+ "join-select rolling backward, in-memory false"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ ",2020/12/12,1"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"],
+ "join-select rolling backward, in-memory true"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ ",2020/12/12,1"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/05,2020/12/01,2"],
+ "join-select rolling forward, in-memory false"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/01,2"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/10,2020/12/12,1"],
+ "join-select rolling forward, in-memory true"
+ ["2_UpdateDate,1_UpdateDate,1_Employee"
+ ",2020/12/01,2"
+ ",2020/12/03,5"
+ ",2020/12/05,4"
+ ",2020/12/05,6"
+ "2020/12/03,2020/12/03,3"
+ "2020/12/10,2020/12/12,1"]}
diff --git a/test/clojask/engine_test.clj b/test/clojask/engine_test.clj
index a640060..95d6382 100644
--- a/test/clojask/engine_test.clj
+++ b/test/clojask/engine_test.clj
@@ -1,85 +1,79 @@
(ns engine-test
- "The same computes on the local engine and on Onyx. While the local
- engine runs, starting Onyx fails the test. Starts Onyx."
+ "Every kind of compute against the output Onyx wrote for it before
+ clojask 2.1 removed Onyx, recorded in correct_outputs/engine.edn: per
+ case the header and the sorted rows."
(:require [clojure.test :refer :all]
+ [clojure.edn :as edn]
[clojure.java.io :as io]
[clojask.dataframe :as ck]
- [clojask.onyx-comps :as oc]
[clojask.api.aggregate :as agg]
[clojask.api.gb-aggregate :as gb])
(:import [com.clojask.exception ExecutionException]))
(def input "test/clojask/Employees-example.csv")
+(def info "test/clojask/Employees-info-example.csv")
(def dir "test/clojask/test_outputs/engine/")
-(defn- on-engine
- [engine f]
- (let [old (System/getProperty "clojask.engine")]
- (System/setProperty "clojask.engine" engine)
- (try
- (if (= engine "local")
- (with-redefs [oc/with-onyx-env (fn [& _] (throw (AssertionError. "the local engine started Onyx")))]
- (f))
- (f))
- (finally
- (if old
- (System/setProperty "clojask.engine" old)
- (System/clearProperty "clojask.engine"))))))
+(def ^:private expected
+ (delay (edn/read-string (slurp "test/clojask/correct_outputs/engine.edn"))))
(defn- lines [path]
(with-open [r (io/reader path)] (vec (line-seq r))))
(defn- result
"The header and the sorted rows that compute writes for the dataframe
- (make-df) on engine, to a file or, with :in-memory, to the group store."
- [engine make-df & opts]
- (on-engine engine
- (fn []
- (let [path (str dir engine ".csv")]
- (io/make-parents path)
- (apply ck/compute (make-df) 2 path opts)
- (let [[header & rows] (lines path)]
- (into [header] (sort rows)))))))
-
-(defn- same-on-both
+ (make-df)."
[make-df & opts]
- (let [onyx (apply result "onyx" make-df opts)]
- (is (< 1 (count onyx)) "the computed frame has rows")
- (is (= onyx (apply result "local" make-df opts)))))
+ (let [path (str dir "result.csv")]
+ (io/make-parents path)
+ (apply ck/compute (make-df) 2 path opts)
+ (let [[header & rows] (lines path)]
+ (into [header] (sort rows)))))
+
+(defn- matches
+ "Compute (make-df) with opts and compare with the recorded output of
+ case k."
+ [k make-df & opts]
+ (let [want (get @expected k)]
+ (is (< 1 (count want)) (str "case " k " is recorded with rows"))
+ (is (= want (apply result make-df opts)) k)))
(defn- salaries []
(doto (ck/dataframe input)
(ck/set-type "Salary" "double")))
(deftest aggregate-only
- (same-on-both #(doto (salaries)
- (ck/filter "Salary" (fn [s] (<= s 800)))
- (ck/aggregate agg/max ["Salary"] ["max"])
- (ck/aggregate agg/min ["Salary"] ["min"])
- (ck/aggregate agg/count ["Employee"] ["n"]))))
+ (matches "aggregate-only"
+ #(doto (salaries)
+ (ck/filter "Salary" (fn [s] (<= s 800)))
+ (ck/aggregate agg/max ["Salary"] ["max"])
+ (ck/aggregate agg/min ["Salary"] ["min"])
+ (ck/aggregate agg/count ["Employee"] ["n"]))))
(deftest group-aggregate
(doseq [in-memory [false true]]
- (testing (str "in-memory " in-memory)
- (same-on-both #(doto (salaries)
- (ck/group-by ["Department"])
- (ck/aggregate gb/max ["Salary"] ["max"])
- (ck/aggregate gb/mean ["Salary"] ["mean"]))
- :in-memory in-memory))))
+ (matches (str "group-aggregate, in-memory " in-memory)
+ #(doto (salaries)
+ (ck/group-by ["Department"])
+ (ck/aggregate gb/max ["Salary"] ["max"])
+ (ck/aggregate gb/mean ["Salary"] ["mean"]))
+ :in-memory in-memory)))
(deftest group-aggregate-multi-row-result
- (same-on-both #(doto (salaries)
- (ck/group-by ["Department"])
- (ck/aggregate gb/smallest3 ["Salary"] ["low"]))
- :select ["low" "Department"]))
+ (matches "group-aggregate-multi-row-result"
+ #(doto (salaries)
+ (ck/group-by ["Department"])
+ (ck/aggregate gb/smallest3 ["Salary"] ["low"]))
+ :select ["low" "Department"]))
(deftest group-by-only
- (same-on-both #(doto (ck/dataframe input)
- (ck/group-by ["Department"]))))
+ (matches "group-by-only"
+ #(doto (ck/dataframe input)
+ (ck/group-by ["Department"]))))
(def ^:private boom?
"Set once the dataframe is built, so the preview that checks each
- aggregate does not throw."
+ function does not throw."
(atom false))
(defn- boom [f]
@@ -88,24 +82,27 @@
(throw (RuntimeException. "boom"))
(apply f args))))
-(deftest aggregate-failure-is-staged
- (doseq [engine ["onyx" "local"]
- [label make-df] [["aggregate" #(ck/aggregate (salaries) (boom agg/max) ["Salary"] ["x"])]
- ["group aggregate" #(doto (salaries)
- (ck/group-by ["Department"])
- (ck/aggregate (boom gb/max) ["Salary"] ["x"]))]]]
- (testing (str engine " " label)
- (let [df (make-df)]
- (reset! boom? true)
- (try
- (is (thrown-with-msg? ExecutionException #"boom"
- (on-engine engine #(ck/compute df 2 (str dir "failed.csv")))))
- (finally
- (reset! boom? false))))
- (is (empty? (rest (file-seq (io/file ".clojask/grouped"))))
- "the group files are deleted"))))
+(defn- fails-staged
+ "Computing df throws an ExecutionException about boom, and the group
+ files are gone afterwards."
+ [df & opts]
+ (reset! boom? true)
+ (try
+ (is (thrown-with-msg? ExecutionException #"boom"
+ (apply ck/compute df 2 (str dir "failed.csv") opts)))
+ (finally
+ (reset! boom? false)))
+ (doseq [d [".clojask/grouped" ".clojask/join"]]
+ (is (empty? (remove #(.isDirectory %) (file-seq (io/file d))))
+ (str "the files under " d " are deleted"))))
-(def info "test/clojask/Employees-info-example.csv")
+(deftest aggregate-failure-is-staged
+ (testing "aggregate"
+ (fails-staged (ck/aggregate (salaries) (boom agg/max) ["Salary"] ["x"])))
+ (testing "group aggregate"
+ (fails-staged (doto (salaries)
+ (ck/group-by ["Department"])
+ (ck/aggregate (boom gb/max) ["Salary"] ["x"])))))
(defn- employees
"The employees with a filter, an operation and a formatter, so the join
@@ -132,31 +129,23 @@
["rolling forward" #(ck/rolling-join-forward (employees) (infos) ["EmployeeName"] ["EmployeeName"] "UpdateDate" "UpdateDate")]
["rolling backward" #(ck/rolling-join-backward (employees) (infos) ["EmployeeName"] ["EmployeeName"] "UpdateDate" "UpdateDate")]
["rolling backward with limit" #(ck/rolling-join-backward (employees) (infos) ["Employee"] ["Employee"] "UpdateDate" "UpdateDate"
- :limit (fn [_ b-val] (not= b-val "2020/12/05")))]])
+ :limit (fn [_ b-val] (not= b-val "2020/12/05")))]])
(deftest join
(doseq [[label make-df] joins
in-memory [false true]]
- (testing (str label ", in-memory " in-memory)
- (same-on-both make-df :in-memory in-memory))))
+ (matches (str "join " label ", in-memory " in-memory)
+ make-df :in-memory in-memory)))
(deftest join-select
(doseq [[label make-df] joins
in-memory [false true]]
- (testing (str label ", in-memory " in-memory)
- (same-on-both make-df :select ["2_UpdateDate" "1_UpdateDate" "1_Employee"] :in-memory in-memory))))
+ (matches (str "join-select " label ", in-memory " in-memory)
+ make-df :select ["2_UpdateDate" "1_UpdateDate" "1_Employee"] :in-memory in-memory)))
(deftest join-failure-is-staged
- (doseq [engine ["onyx" "local"]
- [label make-df] [["inner" ck/inner-join] ["outer" ck/outer-join]]
+ (doseq [[label make-join] [["inner" ck/inner-join] ["outer" ck/outer-join]]
in-memory [false true]]
- (testing (str engine " " label ", in-memory " in-memory)
- (let [df (make-df (doto (salaries) (ck/operate (boom -) "Salary")) (infos) ["Employee"] ["Employee"])]
- (reset! boom? true)
- (try
- (is (thrown-with-msg? ExecutionException #"boom"
- (on-engine engine #(ck/compute df 2 (str dir "failed.csv") :exception true :in-memory in-memory))))
- (finally
- (reset! boom? false))))
- (is (empty? (remove #(.isDirectory %) (file-seq (io/file ".clojask/join"))))
- "the join files are deleted"))))
+ (testing (str label ", in-memory " in-memory)
+ (fails-staged (make-join (doto (salaries) (ck/operate (boom -) "Salary")) (infos) ["Employee"] ["Employee"])
+ :exception true :in-memory in-memory))))
diff --git a/test/clojask/inmemory_test.clj b/test/clojask/inmemory_test.clj
index 1741e72..aa3a180 100644
--- a/test/clojask/inmemory_test.clj
+++ b/test/clojask/inmemory_test.clj
@@ -15,8 +15,9 @@
(defn get-diff
"Compare an output file with the expected one. The header line must match
- exactly; data rows are compared sorted (Onyx writes them in any order)
- unless order is false."
+ exactly; data rows are compared sorted (group-by and outer-join output
+ comes in group order, which depends on the file system) unless order
+ is false."
[a b & [order]]
(try
(let [order (if (nil? order) true order)
diff --git a/test/clojask/regression_test.clj b/test/clojask/regression_test.clj
index 00f53f8..8de3cd8 100644
--- a/test/clojask/regression_test.clj
+++ b/test/clojask/regression_test.clj
@@ -1,5 +1,5 @@
(ns regression-test
- "Computes that each exercised a bug fixed in 2.0.5. Starts Onyx."
+ "Computes that each exercised a bug fixed in 2.0.5."
(:require [clojure.test :refer :all]
[clojure.java.io :as io]
[clojure.string :as str]
@@ -115,3 +115,18 @@
(is (= 22 (count (lines outer))) "header, one match, twenty unmatched b rows")
(ck/compute (ck/inner-join a b ["k"] ["k"]) 1 selected :select ["2_bv" "1_av"] :in-memory in-memory)
(is (= ["2_bv,1_av" "b1,a1"] (lines selected))))))))
+
+(deftest sort-orders-rows-by-the-trending-list
+ (let [df (ck/dataframe input)
+ path (out "sorted.csv")]
+ (ck/set-type df "Salary" "double")
+ (ck/sort df ["+" "Department" "-" "Salary"] path)
+ (is (= ["Employee,EmployeeName,Department,Salary,UpdateDate"
+ "7,Amy,11,50000,2020/11/26"
+ "2,Bob,11,600,2020/12/01"
+ "1,Alice,11,300,2020/12/12"
+ "4,Daniel,12,1000,2020/12/05"
+ "3,Carla,12,900,2020/12/03"
+ "5,Evelyn,13,800,2020/12/03"
+ "6,Ferdinand,21,700,2020/12/05"]
+ (lines path)))))
diff --git a/test/clojask/shutdown_test.clj b/test/clojask/shutdown_test.clj
deleted file mode 100644
index bc6e1c6..0000000
--- a/test/clojask/shutdown_test.clj
+++ /dev/null
@@ -1,107 +0,0 @@
-(ns shutdown-test
- (:require [clojure.test :refer :all]
- [clojure.java.io :as io]
- [clojask.onyx-comps :as oc]
- [clojask.aggregate.aggre-onyx-comps :as ag]
- [clojask.dataframe :as ck]
- [onyx.api]
- [taoensso.timbre :as timbre])
- (:import [com.clojask.exception ExecutionException]))
-
-(def input "test/clojask/Employees-example.csv")
-
-(defn- on-onyx
- "These tests exercise the Onyx environment, whichever engine the rest of
- the suite runs on."
- [t]
- (let [old (System/getProperty "clojask.engine")]
- (System/setProperty "clojask.engine" "onyx")
- (try
- (t)
- (finally
- (if old
- (System/setProperty "clojask.engine" old)
- (System/clearProperty "clojask.engine"))))))
-
-(use-fixtures :each on-onyx)
-
-(defn- env-is-down []
- (is (nil? oc/env))
- (is (nil? oc/peer-group))
- (is (nil? oc/v-peers)))
-
-(defn- compute-still-works []
- (let [out "test/clojask/test_outputs/shutdown-test.csv"
- df (ck/dataframe input)]
- (io/make-parents out)
- (ck/compute df 2 out)
- (is (= 8 (with-open [r (io/reader out)] (count (line-seq r)))))))
-
-(deftest partial-startup-releases-zookeeper
- (testing "A startup that fails after ZooKeeper is up can be shut down, and later computes still work"
- (oc/workflow-gen 1)
- ;; Force the peer group to fail after the env (embedded ZooKeeper) is up.
- (with-redefs [onyx.api/start-peer-group
- (fn [_] (throw (RuntimeException. "simulated peer-group failure")))]
- (is (thrown? RuntimeException (oc/config-env))))
- (is (some? oc/env) "env was started before the failure")
- (is (nil? oc/peer-group) "peer group never started")
- ;; Must not throw, and must release ZooKeeper's port.
- (oc/shutdown)
- (is (nil? oc/env))
- ;; A real compute in the same JVM must now succeed.
- (compute-still-works)))
-
-(deftest failed-job-preparation-releases-environment
- (testing "with-onyx-env tears the environment down when a job's preparation fails"
- (is (thrown-with-msg? ExecutionException #"preparing stage \(probe\)"
- (oc/with-onyx-env "probe" 1
- (fn [] (oc/run-job "probe"
- (fn [] (throw (RuntimeException. "simulated prepare failure"))))))))
- (env-is-down)))
-
-(deftest failed-aggregate-preparation-releases-environment
- (testing "The group-by aggregate driver shares that handling"
- (let [df (ck/dataframe input)]
- (with-redefs [ag/worker-func-gen (fn [& _] (throw (RuntimeException. "simulated prepare failure")))]
- (is (thrown-with-msg? ExecutionException #"preparing stage \(groupby aggregate\)"
- (ag/start-onyx-aggre 1 10 df nil "test/clojask/test_outputs/never.csv" false [] [] {} nil))))
- (env-is-down)
- (compute-still-works))))
-
-(deftest ports-and-directory-are-configurable
- (testing "system properties choose the ZooKeeper port, the Aeron port and the Aeron directory"
- (let [dir (str (System/getProperty "java.io.tmpdir") "/clojask-test-aeron")]
- (try
- (System/setProperty "clojask.zookeeper.port" "2199")
- (System/setProperty "clojask.aeron.port" "40211")
- (System/setProperty "clojask.aeron.dir" dir)
- (compute-still-works)
- (is (= 2199 (:zookeeper.server/port oc/env-config)))
- (is (= "127.0.0.1:2199" (:zookeeper/address oc/peer-config)))
- (is (= 40211 (:onyx.messaging/peer-port oc/peer-config)))
- (is (= dir (:onyx.messaging.aeron/media-driver-dir oc/peer-config)))
- (finally
- (System/clearProperty "clojask.zookeeper.port")
- (System/clearProperty "clojask.aeron.port")
- (System/clearProperty "clojask.aeron.dir"))))
- (compute-still-works)
- (is (= 2188 (:zookeeper.server/port oc/env-config)) "defaults are back")
- (is (nil? (:onyx.messaging.aeron/media-driver-dir oc/peer-config)))))
-
-(deftest log-stays-small
- (testing "a compute adds no :info chatter to the Onyx log"
- (let [log (io/file oc/log-path)
- before (if (.exists log) (.length log) 0)]
- (compute-still-works)
- (is (< (- (.length log) before) 20000)
- "Onyx's default :info logging adds a few hundred KB per compute"))))
-
-(deftest compute-keeps-debug-log-level
- (testing "The log configuration handed to Onyx leaves enable-debug in effect"
- (try
- (ck/enable-debug)
- (compute-still-works)
- (is (= :debug (:min-level timbre/*config*)))
- (finally
- (ck/disable-debug)))))