<commit_before>package com.crystal

// Models
import models.User

// Spark
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{ Duration, StreamingContext }
import org.apache.spark.streaming.kinesis._
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream

// JSON Parsing
import scala.util.parsing.json.JSON

object Main extends App {
  AppConfig.setArgs(args)

  AppConfig.load() match {
    case Some(appConfig) =>
      val sparkConf = new SparkConf()
        .setMaster("local[2]")
        .setAppName(appConfig.appName)

      val streamingCtx = new StreamingContext(
        sparkConf,
        Duration(appConfig.checkpointInterval)
      )

      val kinesisStream = KinesisUtils.createStream(
        streamingCtx,
        appConfig.appName,
        appConfig.streamName,
        s"kinesis.${appConfig.regionName}.amazonaws.com",
        appConfig.regionName,
        InitialPositionInStream.LATEST,
        Duration(appConfig.checkpointInterval),
        StorageLevel.MEMORY_AND_DISK_2
      )

      val eventStream = kinesisStream
        .map { byteArray => new String(byteArray) }
        .map { stringVal => JSON.parseFull(stringVal).get.asInstanceOf[Map[String, Any]] }


      val userStream = eventStream
        .map { e => (e.get(appConfig.userIdentifier).get.asInstanceOf[String], Array(e)) }
        .reduceByKey { (a, b) => a.union(b) }
        .map {
          case (user_id, eventList) =>
            eventList.foldLeft(User.withID(user_id)) { (b, a) => b.performedAction(a) }
        }

      userStream.foreachRDD { rdd =>
        rdd.collect().foreach{ user =>
          println(user.id)
          user.save()
        }
      }

      streamingCtx.start()

      streamingCtx
        .awaitTerminationOrTimeout(appConfig.checkpointInterval)

    case None => ()
  }
}<commit_msg>Make mapping a little easier to read<commit_after>package com.crystal

// Models
import models.User

// Spark
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{ Duration, StreamingContext }
import org.apache.spark.streaming.kinesis._
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream

// JSON Parsing
import scala.util.parsing.json.JSON

object Main extends App {
  AppConfig.setArgs(args)

  AppConfig.load() match {
    case Some(appConfig) =>
      val sparkConf = new SparkConf()
        .setMaster("local[2]")
        .setAppName(appConfig.appName)

      val streamingCtx = new StreamingContext(
        sparkConf,
        Duration(appConfig.checkpointInterval)
      )

      val kinesisStream = KinesisUtils.createStream(
        streamingCtx,
        appConfig.appName,
        appConfig.streamName,
        s"kinesis.${appConfig.regionName}.amazonaws.com",
        appConfig.regionName,
        InitialPositionInStream.LATEST,
        Duration(appConfig.checkpointInterval),
        StorageLevel.MEMORY_AND_DISK_2
      )

      val eventStream = kinesisStream
        .map { byteArray => new String(byteArray) }
        .map { stringVal => JSON.parseFull(stringVal).get.asInstanceOf[Map[String, Any]] }


      val userStream = eventStream
        .map { e =>
          val user_id = e.get(appConfig.userIdentifier).get.asInstanceOf[String]

          (user_id, Array(e))
        }
        .reduceByKey { (a, b) => a.union(b) }
        .map {
          case (user_id, eventList) =>
            eventList.foldLeft(User.withID(user_id)) { (b, a) => b.performedAction(a) }
        }

      userStream.foreachRDD { rdd =>
        rdd.collect().foreach{ user =>
          println(user.id)
          user.save()
        }
      }

      streamingCtx.start()

      streamingCtx
        .awaitTerminationOrTimeout(appConfig.checkpointInterval)

    case None => ()
  }
}
{'fraction_non_alphanumeric': 0.10913312693498452, 'fraction_numerical': 0.0010319917440660474, 'mean_word_length': 2.6233644859813086, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, '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': '2633342', 'n_tokens_mistral': 1213, 'n_tokens_neox': 1186, 'n_words': 199}