onyx learn kit 6.1 challenge

[Punit Naik]

I am facing a problem while executing the 6.1 challenge from the 'learn-onyx' kit.

I changed the :window/aggregation key to have the value of [:onyx.windowing.aggregation/sum :bytes-sent] but I am getting an error like this:

java.lang.IllegalArgumentException: Don't know how to create ISeq from: java.lang.Long

Could anyone please tell me what's wrong?

[Lucas Bradstreet]

Hi Punit,

Could you please post the full window map (containing the
:window/aggregation-key) that is being submitted?
Lucas

[Punit Naik]

Here is the full source:

(ns workshop.challenge-6-1
  (:require [workshop.workshop-utils :as u]))


(def workflow
  [[:read-segments :bucket-page-views]])


(defn build-catalog
  ([] (build-catalog 5 50))
  ([batch-size batch-timeout]
     [{:onyx/name :read-segments
       :onyx/plugin :onyx.plugin.core-async/input
       :onyx/type :input
       :onyx/medium :core.async
       :onyx/batch-size batch-size
       :onyx/batch-timeout batch-timeout
       :onyx/max-peers 1
       :onyx/doc "Reads segments from a core.async channel"}


      {:onyx/name :bucket-page-views
       :onyx/plugin :onyx.peer.function/function
       :onyx/fn :clojure.core/identity
       :onyx/type :output
       :onyx/medium :function
       :onyx/uniqueness-key :event-id
       :onyx/batch-size batch-size
       :onyx/batch-timeout batch-timeout
       :onyx/doc "Identity function, used for windowing segments unchanged."}
      ]))


(defn inject-reader-ch [event lifecycle]
  {:core.async/chan (u/get-input-channel (:core.async/id lifecycle))})


(def reader-lifecycle
  {:lifecycle/before-task-start inject-reader-ch})


(defn build-lifecycles []
  [{:lifecycle/task :read-segments
    :lifecycle/calls :workshop.challenge-6-1/reader-lifecycle
    :core.async/id (java.util.UUID/randomUUID)
    :onyx/doc "Injects the core.async reader channel"}


   {:lifecycle/task :read-segments
    :lifecycle/calls :onyx.plugin.core-async/reader-calls
    :onyx/doc "core.async plugin base lifecycle"}])


(def windows
  [{:window/id :collect-segments
    :window/task :bucket-page-views
    :window/type :fixed
    :window/aggregation [:onyx.windowing.aggregation/sum :bytes-sent]
    :window/window-key :event-time
    :window/range [2 :hour]
    :window/doc "Conj's segments into one hour fixed windows."}])


(def triggers
  [{:trigger/window-id :collect-segments
    :trigger/refinement :discarding
    :trigger/on :segment
    :trigger/fire-all-extents? true
    :trigger/threshold [10 :elements]
    :trigger/sync ::deliver-promise!
    :trigger/doc "Fires against all extents every 10 segments processed."}])


(def fired-window-state (atom {}))
(defn deliver-promise! [event window trigger {:keys [window-id lower-bound upper-bound]} state]
  (let [lower (java.util.Date. lower-bound)
        upper (java.util.Date. upper-bound)]
    (swap! fired-window-state assoc [lower upper] (into #{} state))))

[Lucas Bradstreet]

Hi Punit,

If you print state, you'll see that it's a long, therefore (into #{}
state) will throw the error that you're seeing.

This is because the trigger iterates over each window and calls the
sync function each time on the state for each window (which is a long
value in this case).

Lucas

> --
> You received this message because you are subscribed to the Google Groups
> "Onyx" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to onyx-user+...@googlegroups.com.
> To post to this group, send email to onyx...@googlegroups.com.
> To view this discussion on the web visit
> https://groups.google.com/d/msgid/onyx-user/71f77907-3122-4ed5-ac5f-27a35766064a%40googlegroups.com.
>
> For more options, visit https://groups.google.com/d/optout.

[Punit Naik]

Thanks a lot Lucas! I fixed the error (Y) :)