<commit_before>;; ## EXPERIMENTAL
;;
;; This is a partial and mostly untested implementation of
;; Spark Streaming; consider it a work in progress.
;;
(ns flambo.streaming
  (:refer-clojure :exclude [map time print union count])
  (:require [flambo.api :as f]
            [flambo.conf :as conf]
            [flambo.function :refer [flat-map-function
                                     function
                                     function2
                                     pair-function
                                     void-function]])
  (:import [org.apache.spark.streaming.api.java JavaStreamingContext]
           [org.apache.spark.streaming.kafka KafkaUtils]
           [org.apache.spark.streaming Duration Time]))

(defn duration [ms]
  (Duration. ms))

(defn time [ms]
  (Time. ms))

(defn streaming-context 
  "conf can be a SparkConf or JavaSparkContext"
  [conf batch-duration]
  (JavaStreamingContext. conf (duration batch-duration)))

(defn local-streaming-context [app-name duration]
  (let [conf (-> (conf/spark-conf)
                 (conf/master "local")
                 (conf/app-name app-name))]
    (streaming-context conf duration)))

(defmulti checkpoint (fn [context arg] (class arg)))
(defmethod checkpoint java.lang.String [streaming-context path] (.checkpoint streaming-context path))
(defmethod checkpoint java.lang.Long [dstream interval] (.checkpoint dstream (duration interval)))

(defn text-file-stream [context file-path]
  (.textFileStream context file-path))

(defn socket-text-stream [context ip port]
  (.socketTextStream context ip port))

(defn kafka-stream [& {:keys [streaming-context zk-connect group-id topic-map]}]
  (KafkaUtils/createStream streaming-context zk-connect group-id (into {} (for [[k, v] topic-map] [k (Integer. v)]))))

(defn flat-map [dstream f]
  (.flatMap dstream (flat-map-function f)))

(defn map [dstream f]
  (.map dstream (function f)))

(defn reduce-by-key [dstream f]
  (-> dstream
      (.mapToPair (pair-function identity))
      (.reduceByKey (function2 f))
      (.map (function f/untuple))))<commit_msg>Check the type of dstream before doing reduceByKey.<commit_after>;; ## EXPERIMENTAL
;;
;; This is a partial and mostly untested implementation of
;; Spark Streaming; consider it a work in progress.
;;
(ns flambo.streaming
  (:refer-clojure :exclude [map time print union count])
  (:require [flambo.api :as f]
            [flambo.conf :as conf]
            [flambo.function :refer [flat-map-function
                                     function
                                     function2
                                     pair-function
                                     void-function]])
  (:import [org.apache.spark.streaming.api.java JavaStreamingContext]
           [org.apache.spark.streaming.kafka KafkaUtils]
           [org.apache.spark.streaming Duration Time]))

(defn duration [ms]
  (Duration. ms))

(defn time [ms]
  (Time. ms))

(defn streaming-context 
  "conf can be a SparkConf or JavaSparkContext"
  [conf batch-duration]
  (JavaStreamingContext. conf (duration batch-duration)))

(defn local-streaming-context [app-name duration]
  (let [conf (-> (conf/spark-conf)
                 (conf/master "local")
                 (conf/app-name app-name))]
    (streaming-context conf duration)))

(defmulti checkpoint (fn [context arg] (class arg)))
(defmethod checkpoint java.lang.String [streaming-context path] (.checkpoint streaming-context path))
(defmethod checkpoint java.lang.Long [dstream interval] (.checkpoint dstream (duration interval)))

(defn text-file-stream [context file-path]
  (.textFileStream context file-path))

(defn socket-text-stream [context ip port]
  (.socketTextStream context ip port))

(defn kafka-stream [& {:keys [streaming-context zk-connect group-id topic-map]}]
  (KafkaUtils/createStream streaming-context zk-connect group-id (into {} (for [[k, v] topic-map] [k (Integer. v)]))))

(defn flat-map [dstream f]
  (.flatMap dstream (flat-map-function f)))

(defn map [dstream f]
  (.map dstream (function f)))

(defn reduce-by-key [dstream f]
  "Call reduceByKey on dstream of type JavaDStream or JavaPairDStream"
  (if (instance? org.apache.spark.streaming.api.java.JavaDStream dstream)
    ;; JavaDStream doesn't have a .reduceByKey so cast to JavaPairDStream first
    (-> dstream
      (.mapToPair (pair-function identity))
      (.reduceByKey (function2 f))
      (.map (function f/untuple)))
    ;; if it's already JavaPairDStream, we're good
    (-> dstream
        (.reduceByKey (function2 f))
        (.map (function f/untuple)))))
{'fraction_non_alphanumeric': 0.12399039511023793, 'fraction_numerical': 0.0010914647456887142, 'mean_word_length': 2.796188898094449, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5264853', 'n_tokens_mistral': 1434, 'n_tokens_neox': 1365, 'n_words': 398}