<commit_before>package ws.vinta.albedo

import org.apache.spark.SparkConf
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SparkSession}
import ws.vinta.albedo.utils.DatasetUtils._

import scala.collection.mutable

object RepoProfileBuilder {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    if (scala.util.Properties.envOrElse("RUN_WITH_INTELLIJ", "false") == "true") {
      conf.setMaster("local[*]")
      conf.set("spark.driver.memory", "12g")
      //conf.setMaster("local-cluster[1, 3, 12288]")
      //conf.setJars(List("target/albedo-1.0.0-SNAPSHOT-uber.jar"))
      //conf.setMaster("spark://localhost:7077")
      //conf.set("spark.driver.memory", "2g")
      //conf.set("spark.executor.cores", "3")
      //conf.set("spark.executor.memory", "12g")
      //conf.setJars(List("target/albedo-1.0.0-SNAPSHOT-uber.jar"))
    }

    implicit val spark: SparkSession = SparkSession
      .builder()
      .appName("RepoProfileBuilder")
      .config(conf)
      .getOrCreate()

    import spark.implicits._

    // Load Data

    val rawRepoInfoDS = loadRawRepoInfoDS().cache()

    val rawStarringDS = loadRawStarringDS().cache()

    // Feature Engineering

    val continuousColumnNames = mutable.ArrayBuffer.empty[String]
    val categoricalColumnNames = mutable.ArrayBuffer.empty[String]
    val listColumnNames = mutable.ArrayBuffer.empty[String]
    val textColumnNames = mutable.ArrayBuffer.empty[String]

    continuousColumnNames += "repo_size"
    continuousColumnNames += "repo_stargazers_count"
    continuousColumnNames += "repo_forks_count"
    continuousColumnNames += "repo_subscribers_count"
    continuousColumnNames += "repo_open_issues_count"

    categoricalColumnNames += "repo_owner_type"

    // Impute Data

    val nullableColumnNames = Array("repo_description", "repo_homepage")

    val imputedRepoInfoDF = rawRepoInfoDS
      .withColumn("repo_has_null", when(nullableColumnNames.map(rawRepoInfoDS(_).isNull).reduce(_ || _), 1.0).otherwise(0.0))
      .na.fill("", nullableColumnNames)
      .cache()

    categoricalColumnNames += "repo_has_null"

    // Clean Data

    val reducedRepoInfo = imputedRepoInfoDF
      .where($"repo_is_fork" === false)
      .where($"repo_forks_count" <= 90000)
      .where($"repo_stargazers_count".between(30, 100000))
      .cache()

    val lowerableColumnNames = Array("repo_description", "repo_language", "repo_topics")
    val booleanColumnNames = Array("repo_has_issues", "repo_has_projects", "repo_has_downloads", "repo_has_wiki", "repo_has_pages")

    val unmaintainedWords = Array("%unmaintained%", "%no longer maintained%", "%no longer actively maintained%", "%not maintained%", "%not actively maintained%", "%deprecated%", "%moved to%")
    val assignmentWords = Array("%assignment%")

    val cleanRepoInfoDF = (lowerableColumnNames ++ booleanColumnNames).foldLeft[DataFrame](reducedRepoInfo)((accDF, columnName) => {
      columnName match {
        case _ if lowerableColumnNames.contains(columnName) =>
          accDF.withColumn(columnName.replaceFirst("repo_", "repo_clean_"), lower(col(columnName)))
        case _ if booleanColumnNames.contains(columnName) =>
          accDF.withColumn(columnName.replaceFirst("repo_", "repo_clean_"), col(columnName).cast("double"))
      }
    })
    .cache()
    .withColumn("repo_is_unmaintained", when(unmaintainedWords.map($"repo_clean_description".like(_)).reduce(_ or _), 1.0).otherwise(0.0))
    .withColumn("repo_is_assignment", when(assignmentWords.map($"repo_clean_description".like(_)).reduce(_ or _), 1.0).otherwise(0.0))
    .where($"repo_is_unmaintained" === 0 and $"repo_is_assignment" === 0)
    .cache()

    categoricalColumnNames += "repo_clean_has_issues"
    categoricalColumnNames += "repo_clean_has_projects"
    categoricalColumnNames += "repo_clean_has_downloads"
    categoricalColumnNames += "repo_clean_has_wiki"
    categoricalColumnNames += "repo_clean_has_pages"

    // Construct Features

    val vintaStarredRepos = rawStarringDS
      .where($"user_id" === 652070)
      .select($"repo_id".as[Int])
      .collect()
      .to[List]

    val constructedRepoInfoDF = cleanRepoInfoDF
      .withColumn("repo_days_between_created_at_today", datediff(current_date(), $"repo_created_at"))<commit_msg>clean repos<commit_after>package ws.vinta.albedo

import org.apache.spark.SparkConf
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SparkSession}
import ws.vinta.albedo.utils.DatasetUtils._

import scala.collection.mutable

object RepoProfileBuilder {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    if (scala.util.Properties.envOrElse("RUN_WITH_INTELLIJ", "false") == "true") {
      conf.setMaster("local[*]")
      conf.set("spark.driver.memory", "12g")
      //conf.setMaster("local-cluster[1, 3, 12288]")
      //conf.setJars(List("target/albedo-1.0.0-SNAPSHOT-uber.jar"))
      //conf.setMaster("spark://localhost:7077")
      //conf.set("spark.driver.memory", "2g")
      //conf.set("spark.executor.cores", "3")
      //conf.set("spark.executor.memory", "12g")
      //conf.setJars(List("target/albedo-1.0.0-SNAPSHOT-uber.jar"))
    }

    implicit val spark: SparkSession = SparkSession
      .builder()
      .appName("RepoProfileBuilder")
      .config(conf)
      .getOrCreate()

    import spark.implicits._

    // Load Data

    val rawRepoInfoDS = loadRawRepoInfoDS().cache()

    val rawStarringDS = loadRawStarringDS().cache()

    // Feature Engineering

    val continuousColumnNames = mutable.ArrayBuffer.empty[String]
    val categoricalColumnNames = mutable.ArrayBuffer.empty[String]
    val listColumnNames = mutable.ArrayBuffer.empty[String]
    val textColumnNames = mutable.ArrayBuffer.empty[String]

    continuousColumnNames += "repo_size"
    continuousColumnNames += "repo_stargazers_count"
    continuousColumnNames += "repo_forks_count"
    continuousColumnNames += "repo_subscribers_count"
    continuousColumnNames += "repo_open_issues_count"

    categoricalColumnNames += "repo_owner_type"

    // Impute Data

    val nullableColumnNames = Array("repo_description", "repo_homepage")

    val imputedRepoInfoDF = rawRepoInfoDS
      .withColumn("repo_has_null", when(nullableColumnNames.map(rawRepoInfoDS(_).isNull).reduce(_ || _), 1.0).otherwise(0.0))
      .na.fill("", nullableColumnNames)
      .cache()

    categoricalColumnNames += "repo_has_null"

    // Clean Data

    val reducedRepoInfo = imputedRepoInfoDF
      .where($"repo_is_fork" === false)
      .where($"repo_forks_count" <= 90000)
      .where($"repo_stargazers_count".between(30, 100000))
      .cache()

    val lowerableColumnNames = Array("repo_description", "repo_language", "repo_topics")
    val booleanColumnNames = Array("repo_has_issues", "repo_has_projects", "repo_has_downloads", "repo_has_wiki", "repo_has_pages")

    val normalizedRepoInfoDF = (lowerableColumnNames ++ booleanColumnNames).foldLeft[DataFrame](reducedRepoInfo)((accDF, columnName) => {
      columnName match {
        case _ if lowerableColumnNames.contains(columnName) =>
          accDF.withColumn(columnName.replaceFirst("repo_", "repo_clean_"), lower(col(columnName)))
        case _ if booleanColumnNames.contains(columnName) =>
          accDF.withColumn(columnName, col(columnName).cast("double"))
      }
    })
    .cache()

    categoricalColumnNames += "repo_has_issues"
    categoricalColumnNames += "repo_has_projects"
    categoricalColumnNames += "repo_has_downloads"
    categoricalColumnNames += "repo_has_wiki"
    categoricalColumnNames += "repo_has_pages"

    val unmaintainedWords = Array("%unmaintained%", "%no longer maintained%", "%no longer actively maintained%", "%not maintained%", "%not actively maintained%", "%deprecated%", "%moved to%")
    val assignmentWords = Array("%assignment%", "%作業%", "%作业%")
    val demoWords = Array("test", "%demo project%")
    val blogWords = Array("my blog")

    val cleanRepoInfoDF = normalizedRepoInfoDF
      .withColumn("repo_is_unmaintained", when(unmaintainedWords.map($"repo_clean_description".like(_)).reduce(_ or _), 1.0).otherwise(0.0))
      .withColumn("repo_is_assignment", when(assignmentWords.map($"repo_clean_description".like(_)).reduce(_ or _), 1.0).otherwise(0.0))
      .withColumn("repo_is_demo", when(demoWords.map($"repo_clean_description".like(_)).reduce(_ or _) and $"repo_stargazers_count" <= 50, 1.0).otherwise(0.0))
      .withColumn("repo_is_blog", when(blogWords.map($"repo_clean_description".like(_)).reduce(_ or _) and $"repo_stargazers_count" <= 50, 1.0).otherwise(0.0))
      .where($"repo_is_unmaintained" === 0.0)
      .where($"repo_is_assignment" === 0.0)
      .where($"repo_is_demo" === 0.0)
      .where($"repo_is_blog" === 0.0)
      .cache()

    // Construct Features

    val vintaStarredRepos = rawStarringDS
      .where($"user_id" === 652070)
      .select($"repo_id".as[Int])
      .collect()
      .to[List]

    val constructedRepoInfoDF = cleanRepoInfoDF
      .withColumn("repo_days_between_created_at_today", datediff(current_date(), $"repo_created_at"))
{'fraction_non_alphanumeric': 0.14285714285714285, 'fraction_numerical': 0.014230979748221127, 'mean_word_length': 4.2566168009205985, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4226369', 'n_tokens_mistral': 3082, 'n_tokens_neox': 2944, 'n_words': 448}