<commit_before>(ns sparkling.core
  "This is the main entry point to sparkling, typically required like `[sparkling.core :as s]`.

  By design, most operations in sparkling are built up via the thread-last macro `->>`.
  Thus, they work exactly like their clojure.core counterparts, making it easy to migrate Clojure-only code to Spark code.

  If you find an RDD operation missing from the api that you'd like to use, pull requests are
  happily accepted!"

  (:refer-clojure :exclude [map reduce first count take distinct filter group-by values partition-by keys])
  (:require [clojure.tools.logging :as log]
            [sparkling.destructuring :as ds]
            [sparkling.function :refer [flat-map-function
                                        flat-map-function2
                                        function
                                        function2
                                        function3
                                        pair-function
                                        pair-flat-map-function
                                        void-function]]
            [sparkling.scalaInterop :as si]
            [sparkling.conf :as conf]
            [sparkling.utils :as u]
            [sparkling.kryo :as k])
  (:import [scala Tuple2]
           [java.util Comparator ArrayList]
           [org.apache.spark.api.java JavaSparkContext StorageLevels
                                      JavaRDD JavaPairRDD JavaDoubleRDD]
           [org.apache.spark HashPartitioner Partitioner]
           [org.apache.spark.rdd PartitionwiseSampledRDD PartitionerAwareUnionRDD]
           [scala.collection JavaConversions]
           [scala.reflect ClassTag$]))


;; sparkling makes extensive use of kryo to serialize and deserialize clojure functions
;; and data structures. Here we ensure that these properties are set so they are inhereted
;; into any `SparkConf` objects that are created.
;;
;; sparkling WILL NOT WORK without enabling kryo serialization in spark!
;;
;(System/setProperty "spark.serializer" "org.apache.spark.serializer.KryoSerializer")
;(System/setProperty "spark.kryo.registrator" "sparkling.serialization.BaseRegistrator")

(def STORAGE-LEVELS {:memory-only           StorageLevels/MEMORY_ONLY
                     :memory-only-ser       StorageLevels/MEMORY_ONLY_SER
                     :memory-and-disk       StorageLevels/MEMORY_AND_DISK
                     :memory-and-disk-ser   StorageLevels/MEMORY_AND_DISK_SER
                     :disk-only             StorageLevels/DISK_ONLY
                     :memory-only-2         StorageLevels/MEMORY_ONLY_2
                     :memory-only-ser-2     StorageLevels/MEMORY_ONLY_SER_2
                     :memory-and-disk-2     StorageLevels/MEMORY_AND_DISK_2
                     :memory-and-disk-ser-2 StorageLevels/MEMORY_AND_DISK_SER_2
                     :disk-only-2           StorageLevels/DISK_ONLY_2})

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; Spark Context handling

(defn spark-context
  "Creates a spark context that loads settings from given configuration object
   or system properties"
  ([conf]
   (log/debug "JavaSparkContext" (conf/to-string conf))
   (JavaSparkContext. conf))
  ([master app-name]
   (log/debug "JavaSparkContext" master app-name)
   (let [conf (-> (conf/spark-conf)
                  (conf/master master)
                  (conf/app-name app-name))]
     (spark-context conf))))

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

(defmacro with-context
  [context-sym conf & body]
  `(let [~context-sym (sparkling.core/spark-context ~conf)]
     (try
       ~@body
       (finally (.stop ~context-sym)))))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; Utils
(declare count-partitions)

(defn jar-of-ns
  [ns]
  (let [clazz (Class/forName (clojure.string/replace (str ns) #"-" "_"))]
    (JavaSparkContext/jarOfClass clazz)))

(defn tuple [key value]
  (Tuple2. key value))

(defn- ftruthy?
  [f]
  (fn [x] (u/truthy? (f x))))



;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; RDD construction
;;
;; Functions for constructing new RDDs
;;
(defn text-file
  "Reads a text file from HDFS, a local file system (available on all nodes),
  or any Hadoop-supported file system URI, and returns it as an `JavaRDD` of Strings."
  [spark-context filename]
  (.textFile spark-context filename))

(defn parallelize
  "Distributes a local collection to form/return an RDD"
  ([spark-context lst] (.parallelize spark-context lst))
  ([spark-context num-slices lst] (.parallelize spark-context lst num-slices)))

(defn parallelize-pairs
  "Distributes a local collection to form/return an RDD"
  ([spark-context lst] (.parallelizePairs spark-context lst))
  ([spark-context num-slices lst] (.parallelizePairs spark-context lst num-slices)))



;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;<commit_msg>added functions into-rdd and into-pair-rdd for a more clojuresque experience (parallelize just does not describe what it does)<commit_after>(ns sparkling.core
  "This is the main entry point to sparkling, typically required like `[sparkling.core :as s]`.

  By design, most operations in sparkling are built up via the thread-last macro `->>`.
  Thus, they work exactly like their clojure.core counterparts, making it easy to migrate Clojure-only code to Spark code.

  If you find an RDD operation missing from the api that you'd like to use, pull requests are
  happily accepted!"

  (:refer-clojure :exclude [map reduce first count take distinct filter group-by values partition-by keys])
  (:require [clojure.tools.logging :as log]
            [sparkling.destructuring :as ds]
            [sparkling.function :refer [flat-map-function
                                        flat-map-function2
                                        function
                                        function2
                                        function3
                                        pair-function
                                        pair-flat-map-function
                                        void-function]]
            [sparkling.scalaInterop :as si]
            [sparkling.conf :as conf]
            [sparkling.utils :as u]
            [sparkling.kryo :as k])
  (:import [scala Tuple2]
           [java.util Comparator ArrayList]
           [org.apache.spark.api.java JavaSparkContext StorageLevels
                                      JavaRDD JavaPairRDD JavaDoubleRDD]
           [org.apache.spark HashPartitioner Partitioner]
           [org.apache.spark.rdd PartitionwiseSampledRDD PartitionerAwareUnionRDD]
           [scala.collection JavaConversions]
           [scala.reflect ClassTag$]))


;; sparkling makes extensive use of kryo to serialize and deserialize clojure functions
;; and data structures. Here we ensure that these properties are set so they are inhereted
;; into any `SparkConf` objects that are created.
;;
;; sparkling WILL NOT WORK without enabling kryo serialization in spark!
;;
;(System/setProperty "spark.serializer" "org.apache.spark.serializer.KryoSerializer")
;(System/setProperty "spark.kryo.registrator" "sparkling.serialization.BaseRegistrator")

(def STORAGE-LEVELS {:memory-only           StorageLevels/MEMORY_ONLY
                     :memory-only-ser       StorageLevels/MEMORY_ONLY_SER
                     :memory-and-disk       StorageLevels/MEMORY_AND_DISK
                     :memory-and-disk-ser   StorageLevels/MEMORY_AND_DISK_SER
                     :disk-only             StorageLevels/DISK_ONLY
                     :memory-only-2         StorageLevels/MEMORY_ONLY_2
                     :memory-only-ser-2     StorageLevels/MEMORY_ONLY_SER_2
                     :memory-and-disk-2     StorageLevels/MEMORY_AND_DISK_2
                     :memory-and-disk-ser-2 StorageLevels/MEMORY_AND_DISK_SER_2
                     :disk-only-2           StorageLevels/DISK_ONLY_2})

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; Spark Context handling

(defn spark-context
  "Creates a spark context that loads settings from given configuration object
   or system properties"
  ([conf]
   (log/debug "JavaSparkContext" (conf/to-string conf))
   (JavaSparkContext. conf))
  ([master app-name]
   (log/debug "JavaSparkContext" master app-name)
   (let [conf (-> (conf/spark-conf)
                  (conf/master master)
                  (conf/app-name app-name))]
     (spark-context conf))))

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

(defmacro with-context
  [context-sym conf & body]
  `(let [~context-sym (sparkling.core/spark-context ~conf)]
     (try
       ~@body
       (finally (.stop ~context-sym)))))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; Utils
(declare count-partitions)

(defn jar-of-ns
  [ns]
  (let [clazz (Class/forName (clojure.string/replace (str ns) #"-" "_"))]
    (JavaSparkContext/jarOfClass clazz)))

(defn tuple [key value]
  (Tuple2. key value))

(defn- ftruthy?
  [f]
  (fn [x] (u/truthy? (f x))))



;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; RDD construction
;;
;; Functions for constructing new RDDs
;;
(defn text-file
  "Reads a text file from HDFS, a local file system (available on all nodes),
  or any Hadoop-supported file system URI, and returns it as an `JavaRDD` of Strings."
  [spark-context filename]
  (.textFile spark-context filename))

(defn into-rdd
  "Distributes a local collection to form/return an RDD"
  ([spark-context lst] (.parallelize spark-context lst))
  ([spark-context num-slices lst] (.parallelize spark-context lst num-slices)))

(defn into-pair-rdd
  "Distributes a local collection to form/return an RDD"
  ([spark-context lst] (.parallelizePairs spark-context lst))
  ([spark-context num-slices lst] (.parallelizePairs spark-context lst num-slices)))

(def parallelize into-rdd )
(def parallelize-pairs into-pair-rdd)

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
{'fraction_non_alphanumeric': 0.1253095591877167, 'fraction_numerical': 0.0029717682020802376, 'mean_word_length': 2.3867829587386784, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4601806', 'n_tokens_mistral': 3157, 'n_tokens_neox': 2900, 'n_words': 793}