(:require [onyx.plugin.seq])Onyx plugin for reading from a seq. The seq will be read from in a way that is approriate for use with lazy-seqs. Therefore, this plugin can be useful for use with datomic.api/datoms calls, slow lazy calculations, line-seq / buffered reading, etc.
This plugin is included with Onyx. You do not need to add it as a separate dependency.
In your peer boot-up namespace:
(:require [onyx.plugin.seq])Catalog entry:
{:onyx/name :in
:onyx/plugin :onyx.plugin.seq/input
:onyx/type :input
:onyx/medium :seq
:seq/checkpoint? true
:onyx/batch-size batch-size
:onyx/max-peers 1
:onyx/doc "Reads segments from seq"}Lifecycle entry:
[{:lifecycle/task :in
:lifecycle/calls :onyx.plugin.seq/reader-calls}]Ensure that the elements of your seq are maps. If they are not, it may help to bundle them into maps by transforming the elements before returning it from the lifecycle.
e.g. (map (fn [a] {:val a}) ["A" "B" "C"])
Your lifecycle puts the seq into the event as :seq/seq. It may instead put a
function of no arguments that returns the seq. The plugin then calls it when the
task starts and again on every recovery, and holds only its position, so segments
it has read are not kept in memory. Use this for inputs that do not fit in memory.
The plugin will checkpoint the state of the read results. In the case of a
virtual peer crash, the new virtual peer will drop from the seq until it has
reached the first non fully acked segment. In order for this process to work,
the seq that is read from be reproducible on restart. If it is not, please
disable checkpointing via :seq/checkpoint?.
;; :seq/seq is a function, so lines that have been read are not kept in memory.
;; Each call opens the file again; all readers are closed when the task stops.
(defn inject-in-reader [event lifecycle]
(let [readers (atom [])]
{:seq/readers readers
:seq/seq (fn []
(let [rdr (FileReader. (:buffered-reader/filename lifecycle))]
(swap! readers conj rdr)
(map (fn [line] {:line line})
(line-seq (BufferedReader. rdr)))))}))
(defn close-reader [event lifecycle]
(run! #(.close %) @(:seq/readers event)))
(def in-calls
{:lifecycle/before-task-start inject-in-reader
:lifecycle/after-task-stop close-reader})
;; lifecycles
(def lifecycles
[{:lifecycle/task :in
:buffered-reader/filename "test-resources/lines.txt"
:lifecycle/calls ::in-calls}
{:lifecycle/task :in
:lifecycle/calls :onyx.plugin.seq/reader-calls}])onyx.tasks.seq/input-file returns a task map and lifecycle that read a file this way.
Can you improve this documentation? These fine people already did:
Matt Buehlmaier & Joaquim TorresEdit on GitHub
cljdoc builds & hosts documentation for Clojure/Script libraries
| Ctrl+k | Jump to recent docs |
| ← | Move to previous article |
| → | Move to next article |
| Ctrl+/ | Jump to the search field |