<commit_before>(ns deltalog.mr
  (:require [deltalog.schema :as schema])
  (:import [org.apache.hadoop.fs Path FileSystem]
           [org.apache.hadoop.conf Configuration]
           [cascading.tap.hadoop Hfs]
           [cascading.avro AvroScheme]
           [cascading.scheme.hadoop TextDelimited TextLine]
           [cascading.tuple Fields]
           [cascading.operation.expression ExpressionFilter]
           [cascading.pipe Pipe Each GroupBy Every]
           [cascading.operation.aggregator Last] ; for learning
           [cascading.flow Flow FlowDef]
           [cascading.flow.hadoop HadoopFlowConnector]
           [cascading.json.operation JSONSplitter JSONFlatten]))

(defn make-fields [fields] (Fields. (into-array fields)))

(defn make-splitter
  "Creates a JSON parser that will extract each field from the input
  JSON object. The resultant fields will have the same names as they do
  in the json object."
  [json-paths]
  (JSONSplitter. (make-fields json-paths) (into-array json-paths)))

; This creates a tap that will read a text file line-by-line.
; The resultant field will be named "line".
(def in-tap (Hfs. (TextLine. (make-fields ["line"])) "example/input.json"))

(def out-tap (Hfs. (AvroScheme. (schema/avro-schema)) "example/output" true))

(defn -main []
  (let [json-paths ["id" "timestamp" "is_delete" "data"]

        splitter-pipe (Each. "json_split" (make-splitter json-paths))
        tail-pipe (-> splitter-pipe
                      ;(Each. (make-fields ["timestamp"])
                      ;       (ExpressionFilter. "timestamp >= 1393542318" Long/TYPE))
                      (GroupBy. (make-fields ["id"])
                                (make-fields ["timestamp"]))
                      (Every. Fields/ALL (Last.) Fields/RESULTS))

        flow-def (-> (FlowDef/flowDef)
                     (.addSource splitter-pipe in-tap)
                     (.addTailSink tail-pipe out-tap))]

    (-> (HadoopFlowConnector.)
        (.connect flow-def)
        (.complete))))<commit_msg>allow specifying a timestamp on the command line<commit_after>(ns deltalog.mr
  (:require [deltalog.schema :as schema])
  (:import [org.apache.hadoop.fs Path FileSystem]
           [org.apache.hadoop.conf Configuration]
           [cascading.tap.hadoop Hfs]
           [cascading.avro AvroScheme]
           [cascading.scheme.hadoop TextDelimited TextLine]
           [cascading.tuple Fields]
           [cascading.operation.expression ExpressionFilter]
           [cascading.pipe Pipe Each GroupBy Every]
           [cascading.operation.aggregator Last] ; for learning
           [cascading.flow Flow FlowDef]
           [cascading.flow.hadoop HadoopFlowConnector]
           [cascading.json.operation JSONSplitter JSONFlatten]))

(defn make-fields [fields] (Fields. (into-array fields)))

(defn make-splitter
  "Creates a JSON parser that will extract each field from the input
  JSON object. The resultant fields will have the same names as they do
  in the json object."
  [json-paths]
  (JSONSplitter. (make-fields json-paths) (into-array json-paths)))

; This creates a tap that will read a text file line-by-line.
; The resultant field will be named "line".
(def in-tap (Hfs. (TextLine. (make-fields ["line"])) "example/input.json"))

(def out-tap (Hfs. (AvroScheme. (schema/avro-schema)) "example/output" true))

(defn filter-ts-if
  "If timestamp evaluates to true, then this will create a new
  pipe that filters where the 'timestamp' field is greater than or equal
  to the given timestamp value. Otherwise it returns pipe."
  [pipe timestamp]
  (if timestamp
    (Each. pipe
           (make-fields ["timestamp"])
           (ExpressionFilter. (str "timestamp >= " timestamp) Long/TYPE))
    pipe))

(defn -main
  [& args]
  (let [json-paths ["id" "timestamp" "is_delete" "data"]

        splitter-pipe (Each. "json_split" (make-splitter json-paths))
        tail-pipe (-> splitter-pipe
                      (filter-ts-if (first args))
                      (GroupBy. (make-fields ["id"])
                                (make-fields ["timestamp"]))
                      (Every. Fields/ALL (Last.) Fields/RESULTS))

        flow-def (-> (FlowDef/flowDef)
                     (.addSource splitter-pipe in-tap)
                     (.addTailSink tail-pipe out-tap))]

    (-> (HadoopFlowConnector.)
        (.connect flow-def)
        (.complete))))
{'fraction_non_alphanumeric': 0.11995438996579247, 'fraction_numerical': 0.002280501710376283, 'mean_word_length': 2.3947368421052633, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4583710', 'n_tokens_mistral': 1336, 'n_tokens_neox': 1273, 'n_words': 367}