<commit_before>(ns movie-lens.dataset
  "Functions for parsing and loading the MovieLens dataset source files."
  (:require
    [clojure.data.csv :as csv]
    [clojure.java.io :as io]
    [clojure.tools.logging :as log]
    [merkle-db.database :as db]
    [merkle-db.spark.load :as msl]
    [merkle-db.table :as table]
    [merkledag.node :as node]
    [movie-lens.movie :as movie]
    [movie-lens.rating :as rating]
    [movie-lens.tag :as tag]
    [movie-lens.util :as u]
    [multihash.core :as multihash]
    [sparkling.core :as spark]
    [sparkling.destructuring :as sde]))


(defn- csv-rdd
  "Create an RDD from the given CSV file."
  [spark-ctx parts csv-file header parser]
  (->> (spark/text-file spark-ctx (str csv-file) parts)
       (spark/filter (fn remove-header [line] (not= line header)))
       (spark/flat-map csv/read-csv)
       (spark/map parser)))


(defn- load-movies-table!
  [spark-ctx init-store dataset-dir]
  (let [start (System/currentTimeMillis)
        movies-path (str dataset-dir "movies.csv")
        links-path (str dataset-dir "links.csv")]
    (log/info "Loading movies data from" movies-path "and" links-path)
    (let [movies-csv (csv-rdd spark-ctx 4
                              movies-path
                              movie/movies-csv-header
                              movie/parse-movies-row)
          links-csv (csv-rdd spark-ctx 4
                             links-path
                             movie/links-csv-header
                             movie/parse-links-row)
          record-rdd (->> (spark/left-outer-join
                            (spark/key-by ::movie/id movies-csv)
                            (spark/key-by ::movie/id links-csv))
                          (spark/values)
                          (spark/map
                            (sde/fn [(movie ?links)]
                              (merge movie links))))
          table (msl/build-table!
                  init-store
                  movie/table-parameters
                  record-rdd)
          stats (table/collect-stats table)
          elapsed (/ (- (System/currentTimeMillis) start) 1e3)]
      ;(u/pprint table)
      (table/print-stats stats)
      (log/infof "Loaded movies table data (%s) in %s"
                 (multihash/base58 (::node/id table))
                 (u/duration-str elapsed))
      table)))


(defn- load-tags-table!
  [spark-ctx init-store dataset-dir]
  (let [start (System/currentTimeMillis)
        csv-path (str dataset-dir "tags.csv")]
    (log/info "Loading tags data from" csv-path)
    (let [table (msl/build-table!
                  init-store
                  tag/table-parameters
                  (csv-rdd spark-ctx 8
                           csv-path
                           tag/csv-header
                           tag/parse-row))
          stats (table/collect-stats table)
          elapsed (/ (- (System/currentTimeMillis) start) 1e3)]
      (table/print-stats stats)
      (log/infof "Loaded tags table (%s) in %s"
                 (multihash/base58 (::node/id table))
                 (u/duration-str elapsed))
      table)))


(defn- load-ratings-table!
  [spark-ctx init-store dataset-dir]
  (let [start (System/currentTimeMillis)
        csv-path (str dataset-dir "ratings.csv")]
    (log/info "Loading ratings data from" csv-path)
    (let [table (msl/build-table!
                  init-store
                  rating/table-parameters
                  (csv-rdd spark-ctx 64
                           csv-path
                           rating/csv-header
                           rating/parse-row))
          stats (table/collect-stats table)
          elapsed (/ (- (System/currentTimeMillis) start) 1e3)]
      (table/print-stats stats)
      (log/infof "Loaded ratings table (%s) in %s"
                 (multihash/base58 (::node/id table))
                 (u/duration-str elapsed))
      table)))


(defn load-dataset!
  [spark-ctx init-store dataset-dir]
  (let [store (init-store)
        movies (load-movies-table! spark-ctx init-store dataset-dir)
        tags (load-tags-table! spark-ctx init-store dataset-dir)
        ratings (load-ratings-table! spark-ctx init-store dataset-dir)]
    (->
      (db/empty-db store {:data/title "MovieLens Dataset"})
      (db/set-table "movies" movies)
      (db/set-table "ratings" ratings)
      (db/set-table "tags" tags)
      (db/flush!))))<commit_msg>Update example reference to table stats.<commit_after>(ns movie-lens.dataset
  "Functions for parsing and loading the MovieLens dataset source files."
  (:require
    [clojure.data.csv :as csv]
    [clojure.java.io :as io]
    [clojure.tools.logging :as log]
    [merkle-db.database :as db]
    [merkle-db.spark.load :as msl]
    [merkle-db.table :as table]
    [merkle-db.tools.stats :as stats]
    [merkledag.node :as node]
    [movie-lens.movie :as movie]
    [movie-lens.rating :as rating]
    [movie-lens.tag :as tag]
    [movie-lens.util :as u]
    [multihash.core :as multihash]
    [sparkling.core :as spark]
    [sparkling.destructuring :as sde]))


(defn- csv-rdd
  "Create an RDD from the given CSV file."
  [spark-ctx parts csv-file header parser]
  (->> (spark/text-file spark-ctx (str csv-file) parts)
       (spark/filter (fn remove-header [line] (not= line header)))
       (spark/flat-map csv/read-csv)
       (spark/map parser)))


(defn- load-movies-table!
  [spark-ctx init-store dataset-dir]
  (let [start (System/currentTimeMillis)
        movies-path (str dataset-dir "movies.csv")
        links-path (str dataset-dir "links.csv")]
    (log/info "Loading movies data from" movies-path "and" links-path)
    (let [movies-csv (csv-rdd spark-ctx 4
                              movies-path
                              movie/movies-csv-header
                              movie/parse-movies-row)
          links-csv (csv-rdd spark-ctx 4
                             links-path
                             movie/links-csv-header
                             movie/parse-links-row)
          record-rdd (->> (spark/left-outer-join
                            (spark/key-by ::movie/id movies-csv)
                            (spark/key-by ::movie/id links-csv))
                          (spark/values)
                          (spark/map
                            (sde/fn [(movie ?links)]
                              (merge movie links))))
          table (msl/build-table!
                  init-store
                  movie/table-parameters
                  record-rdd)
          stats (stats/collect-table-stats table)
          elapsed (/ (- (System/currentTimeMillis) start) 1e3)]
      ;(u/pprint table)
      (stats/print-table-stats stats)
      (log/infof "Loaded movies table data (%s) in %s"
                 (multihash/base58 (::node/id table))
                 (u/duration-str elapsed))
      table)))


(defn- load-tags-table!
  [spark-ctx init-store dataset-dir]
  (let [start (System/currentTimeMillis)
        csv-path (str dataset-dir "tags.csv")]
    (log/info "Loading tags data from" csv-path)
    (let [table (msl/build-table!
                  init-store
                  tag/table-parameters
                  (csv-rdd spark-ctx 8
                           csv-path
                           tag/csv-header
                           tag/parse-row))
          stats (stats/collect-table-stats table)
          elapsed (/ (- (System/currentTimeMillis) start) 1e3)]
      (stats/print-table-stats stats)
      (log/infof "Loaded tags table (%s) in %s"
                 (multihash/base58 (::node/id table))
                 (u/duration-str elapsed))
      table)))


(defn- load-ratings-table!
  [spark-ctx init-store dataset-dir]
  (let [start (System/currentTimeMillis)
        csv-path (str dataset-dir "ratings.csv")]
    (log/info "Loading ratings data from" csv-path)
    (let [table (msl/build-table!
                  init-store
                  rating/table-parameters
                  (csv-rdd spark-ctx 64
                           csv-path
                           rating/csv-header
                           rating/parse-row))
          stats (stats/collect-table-stats table)
          elapsed (/ (- (System/currentTimeMillis) start) 1e3)]
      (stats/print-table-stats stats)
      (log/infof "Loaded ratings table (%s) in %s"
                 (multihash/base58 (::node/id table))
                 (u/duration-str elapsed))
      table)))


(defn load-dataset!
  [spark-ctx init-store dataset-dir]
  (let [store (init-store)
        movies (load-movies-table! spark-ctx init-store dataset-dir)
        tags (load-tags-table! spark-ctx init-store dataset-dir)
        ratings (load-ratings-table! spark-ctx init-store dataset-dir)]
    (->
      (db/empty-db store {:data/title "MovieLens Dataset"})
      (db/set-table "movies" movies)
      (db/set-table "ratings" ratings)
      (db/set-table "tags" tags)
      (db/flush!))))
{'fraction_non_alphanumeric': 0.12344153656070987, 'fraction_numerical': 0.00381893743681905, 'mean_word_length': 1.6257741079327632, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, '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': '925794', 'n_tokens_mistral': 2829, 'n_tokens_neox': 2693, 'n_words': 629}