<commit_before>/**
 * Happy Panda Example for DataFrames. Computes the % of happy pandas. Very contrived.
 */
package com.highperformancespark.examples.dataframe

import org.apache.spark._
//tag::sparkSQLImports[]
import org.apache.spark.sql.{DataFrame, SQLContext, Row}
import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.expressions._
import org.apache.spark.sql.functions._
//end::sparkSQLImports[]

//tag::sparkHiveImports[]
// Additional imports for using HiveContext
import org.apache.spark.sql.hive._
import org.apache.spark.sql.hive.thriftserver._
//end::sparkHiveImports[]

object HappyPandas {
  // create SQLContext with an existing SparkContext
  def sqlContext(sc: SparkContext): SQLContext = {
    //tag::createSQLContext[]
    val sqlContext = new SQLContext(sc)
    // Import the implicits, unlike in core Spark the implicits are defined on the context
    import sqlContext.implicits._
    //end::createSQLContext[]
    sqlContext
  }

  // create HiveContext with an existing SparkContext
  def hiveContext(sc: SparkContext): HiveContext = {
    //tag::createHiveContext[]
    val hiveContext = new HiveContext(sc)
    // Import the implicits, unlike in core Spark the implicits are defined on the context
    import hiveContext.implicits._
    //end::createHiveContext[]
    hiveContext
  }

  // Illustrate loading some JSON data
  def loadDataSimple(sc: SparkContext, sqlCtx: SQLContext, path: String): DataFrame = {
    //tag::loadPandaJSONSimple[]
    val df = sqlCtx.read.json(path)
    //end::loadPandaJSONSimple[]
    //tag::loadPandaJSONComplex[]
    val df2 = sqlCtx.read.format("json").option("samplingRatio", "1.0").load(path)
    //end::loadPandaJSONComplex[]
    val rdd = sc.textFile(path)
    //tag::loadPandaJsonRDD[]
    val df3 = sqlCtx.read.json(rdd)
    //end::loadPandaJSONRDD[]
    df
  }

  //  Here will be some examples on PandaInfo DataFrame

  /**
    * @param place name of place
    * @param pandaType type of pandas in this place
    * @param happyPandas number of happy pandas in this place
    * @param totalPandas total number of pandas in this place
    */
  case class PandaInfo(place: String, pandaType: String, happyPandas: Integer, totalPandas: Integer)

  /**
    * Gets the percentage of happy pandas per place.
    *
    * @param pandaInfo the input DataFrame
    * @return Returns a pair of (place, percentage of happy pandas)
    */
  def happyPandasPercentage(pandaInfo: DataFrame): DataFrame = {
    pandaInfo.select(pandaInfo("place"),
      (pandaInfo("happyPandas") / pandaInfo("totalPandas")).as("percentHappy"))
  }

  //tag::encodePandaType[]
  /**
    * Encodes pandaType to Integer values instead of String values.
    *
    * @param pandaInfo the input DataFrame
    * @return Returns a DataFrame of pandaId and integer value for pandaType.
    */
  def encodePandaType(pandaInfo: DataFrame): DataFrame = {
    pandaInfo.select(pandaInfo("id"),
      (when(pandaInfo("pt") === "giant", 0).
      when(pandaInfo("pt") === "red", 1).
      otherwise(2)).as("encodedType")
    )
  }
  //end::encodePandaType[]

  /**
    * Gets places with happy pandas more than minHappinessBound.
    */
  def minHappyPandas(pandaInfo: DataFrame, minHappyPandas: Int): DataFrame = {
    pandaInfo.filter(pandaInfo("happyPandas") >= minHappyPandas)
  }

  /**
   * Extra the panda info from panda places and compute the squisheness of the panda
   */
  def squishPandaFromPace(pandaPlace: DataFrame): DataFrame = {
    //tag::selectExplode[]
    val pandaInfo = pandaPlace.explode(pandaPlace("pandas")){
      case Row(pandas: Seq[Row]) =>
        pandas.map{
          case Row(id: Long, zip: String, pt: String, happy: Boolean, attrs: Seq[Double]) =>
            RawPanda(id, zip, pt, happy, attrs.toArray)
        }}
    pandaInfo.select((pandaInfo("attributes")(0) / pandaInfo("attributes")(1)).as("murh"))
    //end::selectExplode[]
  }

  /**
    * Find pandas that are sad
    */
  def sadPandas(pandaInfo: DataFrame): DataFrame = {
    //tag::simpleFilter[]
    pandaInfo.filter(pandaInfo("happy") !== true)
    //end::simpleFilter[]
  }

  /**
   * Find pandas that are happy and fuzzier than squishy
   */
  def happyFuzzyPandas(pandaInfo: DataFrame): DataFrame = {
    //tag::complexFilter[]
    pandaInfo.filter(pandaInfo("happy").and(
      pandaInfo("attributes")(0) > pandaInfo("attributes")(1)))
    //end::complexFilter[]
  }
  /**
    * Gets places that contains happy pandas more than unhappy pandas.
    */
  def happyPandasPlaces(pandaInfo: DataFrame): DataFrame = {
    pandaInfo.filter(pandaInfo("happyPandas") >= pandaInfo("totalPandas") / 2)
  }


  /**
   * Remove duplicate pandas by id
   */
  def removeDuplicates(pandas: DataFrame): DataFrame = {
    //tag::dropDuplicatePandaIds[]
    pandas.dropDuplicates(List("id"))
    //end::dropDuplicatePandaIds[]
  }

  /**
    * @param name name of panda
    * @param zip zip code
    * @param pandaSize size of panda in KG
    * @param age age of panda
    */
  case class Pandas(name: String, zip: String, pandaSize: Integer, age: Integer)

  //tag::maxPandaSizePerZip[]
  def maxPandaSizePerZip(pandas: DataFrame): DataFrame = {
    pandas.groupBy(pandas("zip")).max("pandaSize")
  }
  //end::maxPandaSizePerZip[]

  //tag::minMaxPandasSizePerZip[]
  def minMaxPandaSizePerZip(pandas: DataFrame): DataFrame = {<commit_msg>Add a sample of describe<commit_after>/**
 * Happy Panda Example for DataFrames. Computes the % of happy pandas. Very contrived.
 */
package com.highperformancespark.examples.dataframe

import org.apache.spark._
//tag::sparkSQLImports[]
import org.apache.spark.sql.{DataFrame, SQLContext, Row}
import org.apache.spark.sql.catalyst.expressions.aggregate._
import org.apache.spark.sql.expressions._
import org.apache.spark.sql.functions._
//end::sparkSQLImports[]

//tag::sparkHiveImports[]
// Additional imports for using HiveContext
import org.apache.spark.sql.hive._
import org.apache.spark.sql.hive.thriftserver._
//end::sparkHiveImports[]

object HappyPandas {
  // create SQLContext with an existing SparkContext
  def sqlContext(sc: SparkContext): SQLContext = {
    //tag::createSQLContext[]
    val sqlContext = new SQLContext(sc)
    // Import the implicits, unlike in core Spark the implicits are defined on the context
    import sqlContext.implicits._
    //end::createSQLContext[]
    sqlContext
  }

  // create HiveContext with an existing SparkContext
  def hiveContext(sc: SparkContext): HiveContext = {
    //tag::createHiveContext[]
    val hiveContext = new HiveContext(sc)
    // Import the implicits, unlike in core Spark the implicits are defined on the context
    import hiveContext.implicits._
    //end::createHiveContext[]
    hiveContext
  }

  // Illustrate loading some JSON data
  def loadDataSimple(sc: SparkContext, sqlCtx: SQLContext, path: String): DataFrame = {
    //tag::loadPandaJSONSimple[]
    val df = sqlCtx.read.json(path)
    //end::loadPandaJSONSimple[]
    //tag::loadPandaJSONComplex[]
    val df2 = sqlCtx.read.format("json").option("samplingRatio", "1.0").load(path)
    //end::loadPandaJSONComplex[]
    val rdd = sc.textFile(path)
    //tag::loadPandaJsonRDD[]
    val df3 = sqlCtx.read.json(rdd)
    //end::loadPandaJSONRDD[]
    df
  }

  //  Here will be some examples on PandaInfo DataFrame

  /**
    * @param place name of place
    * @param pandaType type of pandas in this place
    * @param happyPandas number of happy pandas in this place
    * @param totalPandas total number of pandas in this place
    */
  case class PandaInfo(place: String, pandaType: String, happyPandas: Integer, totalPandas: Integer)

  /**
    * Gets the percentage of happy pandas per place.
    *
    * @param pandaInfo the input DataFrame
    * @return Returns a pair of (place, percentage of happy pandas)
    */
  def happyPandasPercentage(pandaInfo: DataFrame): DataFrame = {
    pandaInfo.select(pandaInfo("place"),
      (pandaInfo("happyPandas") / pandaInfo("totalPandas")).as("percentHappy"))
  }

  //tag::encodePandaType[]
  /**
    * Encodes pandaType to Integer values instead of String values.
    *
    * @param pandaInfo the input DataFrame
    * @return Returns a DataFrame of pandaId and integer value for pandaType.
    */
  def encodePandaType(pandaInfo: DataFrame): DataFrame = {
    pandaInfo.select(pandaInfo("id"),
      (when(pandaInfo("pt") === "giant", 0).
      when(pandaInfo("pt") === "red", 1).
      otherwise(2)).as("encodedType")
    )
  }
  //end::encodePandaType[]

  /**
    * Gets places with happy pandas more than minHappinessBound.
    */
  def minHappyPandas(pandaInfo: DataFrame, minHappyPandas: Int): DataFrame = {
    pandaInfo.filter(pandaInfo("happyPandas") >= minHappyPandas)
  }

  /**
   * Extra the panda info from panda places and compute the squisheness of the panda
   */
  def squishPandaFromPace(pandaPlace: DataFrame): DataFrame = {
    //tag::selectExplode[]
    val pandaInfo = pandaPlace.explode(pandaPlace("pandas")){
      case Row(pandas: Seq[Row]) =>
        pandas.map{
          case Row(id: Long, zip: String, pt: String, happy: Boolean, attrs: Seq[Double]) =>
            RawPanda(id, zip, pt, happy, attrs.toArray)
        }}
    pandaInfo.select((pandaInfo("attributes")(0) / pandaInfo("attributes")(1)).as("murh"))
    //end::selectExplode[]
  }

  /**
    * Find pandas that are sad
    */
  def sadPandas(pandaInfo: DataFrame): DataFrame = {
    //tag::simpleFilter[]
    pandaInfo.filter(pandaInfo("happy") !== true)
    //end::simpleFilter[]
  }

  /**
   * Find pandas that are happy and fuzzier than squishy
   */
  def happyFuzzyPandas(pandaInfo: DataFrame): DataFrame = {
    //tag::complexFilter[]
    pandaInfo.filter(pandaInfo("happy").and(
      pandaInfo("attributes")(0) > pandaInfo("attributes")(1)))
    //end::complexFilter[]
  }
  /**
    * Gets places that contains happy pandas more than unhappy pandas.
    */
  def happyPandasPlaces(pandaInfo: DataFrame): DataFrame = {
    pandaInfo.filter(pandaInfo("happyPandas") >= pandaInfo("totalPandas") / 2)
  }


  /**
   * Remove duplicate pandas by id
   */
  def removeDuplicates(pandas: DataFrame): DataFrame = {
    //tag::dropDuplicatePandaIds[]
    pandas.dropDuplicates(List("id"))
    //end::dropDuplicatePandaIds[]
  }

  /**
    * @param name name of panda
    * @param zip zip code
    * @param pandaSize size of panda in KG
    * @param age age of panda
    */
  case class Pandas(name: String, zip: String, pandaSize: Integer, age: Integer)

  def describePandas(pandas: DataFrame): DataFrame = {
    //tag::pandaSizeRangeVarDescribe[]
    pandas.describe()
    //end::pandaSizeRangeVarDescribe[]
  }

  //tag::maxPandaSizePerZip[]
  def maxPandaSizePerZip(pandas: DataFrame): DataFrame = {
    pandas.groupBy(pandas("zip")).max("pandaSize")
  }
  //end::maxPandaSizePerZip[]

  //tag::minMaxPandasSizePerZip[]
  def minMaxPandaSizePerZip(pandas: DataFrame): DataFrame = {
{'fraction_non_alphanumeric': 0.12804822358206228, 'fraction_numerical': 0.002191981002831309, 'mean_word_length': 3.560599750104123, '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': '261892', 'n_tokens_mistral': 3686, 'n_tokens_neox': 3558, 'n_words': 857}