Locality-Sensitive Hashing on Spark with Clojure/Flambo
2017-03-30 23:06:26
How often have you come across requirements that demand tasks to be performed repetitively at a defined interval? Yes, I am talking about a scheduler but a simple, yet powerful one that justifies its name- Just schedules. That is what Nolan Scheduler is all about.
Kuldeep, a champion clojurist, wrote the library and it is now an important part of FORMCEPT platform. It schedules all the jobs within the platform and keeps users up-to-date with the job status.
Email Scheduler
Lets take an example of email scheduler that is required to read emails from an email account, say GMail and do the same periodically. This is the classic use case for a scheduler. So, here is how you can schedule your email reader job-
Step-1: Pick the function to schedule
In this case, we can create a simple function that reads all the unread emails from the specified GMail account. Here is my namespace with the function read-email-
[clojure]
(ns fcgmail.core
^{:author "Anuj" :doc "FORMCEPT GMail Reader"}
(:require [clojure-mail.core :as mcore]
[clojure-mail.message :as m]
[clojure.string :as s]
[clojure.tools.logging :as log]))
; GMail Store Connection
(def ^:private gstore (atom nil))
(defn- read-msg
"Reads the message and returns the subject and body"
[msg]
{:subject (msg :subject)
:body (-> (filter
#(and (:content-type %)
(.startsWith (:content-type %) "TEXT/PLAIN"))
(msg :body))
first :body)})
(defn read-email
"Reads unread emails and marks them as read"
[uri email pwd]
(try
(reset! gstore (mcore/gen-store email pwd))
(let [msgs (mcore/unread-messages @gstore :inbox)
fcmsgs (map #(read-msg (m/read-message %)) msgs)]
(doseq [fcmsg fcmsgs]
(log/info (str "Retrieved: " fcmsg))
; Do whatever you want with the message
(catch Exception e (log/error (str "Failed: " (.getMessage e))))
(finally
(do (mcore/mark-all-read @gstore :inbox)
(mcore/close-store @gstore)))))
[/clojure]
It uses clojure-mail project to connect to GMail and read the messages. I will keep that explanation for the next blog but I encourage readers to go ahead and take a look at this project as well.
Step-2: Schedule
Now, comes the most interesting part. This how you can schedule your target function, i.e. read-email for this example-
[clojure]
(ns fcgmail.core
(:require [nolan.core :as n]))
; Create Scheduler
(defonce sc (n/get-mem-scheduler))
; Schedule
(n/add-schedule sc "R//PT30S" #(read-email uri email pwd))
[/clojure]
That is it :-) - Your scheduled function will be called every 30 seconds as per the repeating intervals syntax of ISO 8601. The function add-schedule returns a schedule ID which can be used later to expire a scehdule which stops all further executions and removes it from schedule store as shown below-
[clojure]
(expire sc scid)
; Check expiry status
(expired? sc scid)
; Should return true
[/clojure]
By default, the library comes with built-in in-memory scheduler but you can extend the ScheduleStore protocol to the store of your choice. Please give it a try.
