<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 max min])
  (: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]
            [sparkling.destructuring :as s-de])
  (: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})



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

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

(def tuple s-de/tuple)

(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 whole-text-files
  "Read a directory of text files from HDFS, a local file system (available on all nodes),
  or any Hadoop-supported file system URI. Each file is read as a single record and returned 
  in a key-value pair, where the key is the path of each file, the value is the content of each file."
  ([spark-context filename min-partitions]
   (.wholeTextFiles spark-context filename min-partitions))
  ([spark-context filename]
   (whole-text-files spark-context filename 8)))

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

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

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

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; Transformations
;; Functions for transforming RDDs
;;

;; functions on one rdd
(defn map
  "Returns a new RDD formed by passing each element of the source through the function `f`."
  [f rdd]
  (-> (.map rdd (function f))
      (u/set-auto-name (u/unmangle-fn f))))

(defn map-to-pair
  "Returns a new `JavaPairRDD` of (K, V) pairs by applying `f` to all elements of `rdd`."
  [f rdd]
  (-> (.mapToPair rdd (pair-function f))
      (u/set-auto-name (u/unmangle-fn f))))

(defn map-values [f rdd]
  (-> (.mapValues rdd (function f))
      (u/set-auto-name (u/unmangle-fn f))))

(defn reduce
  "Aggregates the elements of `rdd` using the function `f` (which takes two arguments
  and returns one). The function should be commutative and associative so that it can be
  computed correctly in parallel."
  [f rdd]
  (u/set-auto-name (.reduce rdd (function2 f)) (u/unmangle-fn f)))

(defn values
  "Returns the values of a JavaPairRDD"
  [rdd]
  (u/set-auto-name (.values rdd)))

(defn flat-map
  "Similar to `map`, but each input item can be mapped to 0 or more output items (so the
   function `f` should return a collection rather than a single item)"
  [f rdd]
  (u/set-auto-name
    (.flatMap rdd (flat-map-function f))
    (u/unmangle-fn f)))

(defn flat-map-to-pair
  "Returns a new `JavaPairRDD` by first applying `f` to all elements of `rdd`, and then flattening
  the results."
  [f rdd]
  (u/set-auto-name
    (.flatMapToPair rdd (pair-flat-map-function f))
    (u/unmangle-fn f)))

(defn flat-map-values
  [f rdd]
  (u/set-auto-name
    (.flatMapValues rdd (function f))
    (u/unmangle-fn f)))

(defn map-partition
  "Similar to `map`, but runs separately on each partition (block) of the `rdd`, so function `f`
  must be of type Iterator<T> => Iterable<U>.
  https://issues.apache.org/jira/browse/SPARK-3369"
  [f rdd]
  (u/set-auto-name
    (.mapPartitions rdd (flat-map-function f))
    (u/unmangle-fn f)))


(defn map-partitions-to-pair
  "Similar to `map`, but runs separately on each partition (block) of the `rdd`, so function `f`
  must be of type Iterator<T> => Iterable<U>.
  https://issues.apache.org/jira/browse/SPARK-3369"
  [f preserve-partitioning? rdd]
  (u/set-auto-name
    (.mapPartitionsToPair rdd (pair-flat-map-function f) (u/truthy? preserve-partitioning?))
    (u/unmangle-fn f)
    preserve-partitioning?))<commit_msg>Added docs to 2 functions<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 max min])
  (: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]
            [sparkling.destructuring :as s-de])
  (: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})



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

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

(def tuple s-de/tuple)

(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 whole-text-files
  "Read a directory of text files from HDFS, a local file system (available on all nodes),
  or any Hadoop-supported file system URI. Each file is read as a single record and returned 
  in a key-value pair, where the key is the path of each file, the value is the content of each file."
  ([spark-context filename min-partitions]
   (.wholeTextFiles spark-context filename min-partitions))
  ([spark-context filename]
   (whole-text-files spark-context filename 8)))

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

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

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

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;
;; Transformations
;; Functions for transforming RDDs
;;

;; functions on one rdd
(defn map
  "Returns a new RDD formed by passing each element of the source through the function `f`."
  [f rdd]
  (-> (.map rdd (function f))
      (u/set-auto-name (u/unmangle-fn f))))

(defn map-to-pair
  "Returns a new `JavaPairRDD` of (K, V) pairs by applying `f` to all elements of `rdd`."
  [f rdd]
  (-> (.mapToPair rdd (pair-function f))
      (u/set-auto-name (u/unmangle-fn f))))

(defn map-values [f rdd]
  "Returns a new `JavaPairRDD` of (K,V) pairs by applying `f` to all values in the 
  pair-rdd `rdd`"  
  (-> (.mapValues rdd (function f))
      (u/set-auto-name (u/unmangle-fn f))))

(defn reduce
  "Aggregates the elements of `rdd` using the function `f` (which takes two arguments
  and returns one). The function should be commutative and associative so that it can be
  computed correctly in parallel."
  [f rdd]
  (u/set-auto-name (.reduce rdd (function2 f)) (u/unmangle-fn f)))

(defn values
  "Returns the values of a JavaPairRDD"
  [rdd]
  (u/set-auto-name (.values rdd)))

(defn flat-map
  "Similar to `map`, but each input item can be mapped to 0 or more output items (so the
   function `f` should return a collection rather than a single item)"
  [f rdd]
  (u/set-auto-name
    (.flatMap rdd (flat-map-function f))
    (u/unmangle-fn f)))

(defn flat-map-to-pair
  "Returns a new `JavaPairRDD` by first applying `f` to all elements of `rdd`, and then flattening
  the results."
  [f rdd]
  (u/set-auto-name
    (.flatMapToPair rdd (pair-flat-map-function f))
    (u/unmangle-fn f)))

(defn flat-map-values
  "Returns a `JavaPairRDD` by applying `f` to all values of `rdd`, and then 
  flattening the results"
  [f rdd]
  (u/set-auto-name
    (.flatMapValues rdd (function f))
    (u/unmangle-fn f)))

(defn map-partition
  "Similar to `map`, but runs separately on each partition (block) of the `rdd`, so function `f`
  must be of type Iterator<T> => Iterable<U>.
  https://issues.apache.org/jira/browse/SPARK-3369"
  [f rdd]
  (u/set-auto-name
    (.mapPartitions rdd (flat-map-function f))
    (u/unmangle-fn f)))


(defn map-partitions-to-pair
  "Similar to `map`, but runs separately on each partition (block) of the `rdd`, so function `f`
  must be of type Iterator<T> => Iterable<U>.
  https://issues.apache.org/jira/browse/SPARK-3369"
  [f preserve-partitioning? rdd]
  (u/set-auto-name
    (.mapPartitionsToPair rdd (pair-flat-map-function f) (u/truthy? preserve-partitioning?))
    (u/unmangle-fn f)
    preserve-partitioning?))
{'fraction_non_alphanumeric': 0.12107189723047172, 'fraction_numerical': 0.0035223427032253607, 'mean_word_length': 2.663040728560587, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 25, 'https://': 4, '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': '6487608', 'n_tokens_mistral': 4834, 'n_tokens_neox': 4545, 'n_words': 1396}