<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.highperformancespark.examples.perf

import com.highperformancespark.examples.dataframe.RawPanda
import com.highperformancespark.examples.tools._

import org.apache.spark.rdd._
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.{DataFrame, Dataset, Row}
import org.apache.spark.sql.hive.HiveContext

/**
 * A simple performance test to compare a simple sort between DataFrame, and RDD
 */
object SimplePerfTest {
  def main(args: Array[String]) = {
    val sparkConf = new SparkConf().setAppName("simple-perf-test")
    val sc = new SparkContext(sparkConf)
    val sqlCtx = new HiveContext(sc)
    val scalingFactor = if (args.length > 0) args(0).toLong else 100L
    val size = if (args.length > 1) args(1).toInt else 50
    run(sc, sqlCtx, scalingFactor, size)
  }

  def run(sc: SparkContext, sqlCtx: HiveContext, scalingFactor: Long, size: Int) = {
    import sqlCtx.implicits._
    val inputRDD = GenerateScalingData.generateFullGoldilocks(sc, scalingFactor, size)
    inputRDD.cache()
    inputRDD.count()
    val rddTimeings = 1.to(10).map(x => time(testOnRDD(inputRDD)))
    val inputDataFrame = inputRDD.toDF()
    inputDataFrame.cache()
    inputDataFrame.count()
    val dataFrameTimeings = 1.to(10).map(x => time(testOnDataFrame(inputDataFrame)))
    println(rddTimeings.map(_._2).mkString(","))
    println(dataFrameTimeings.map(_._2).mkString(","))
  }

  def testOnRDD(rdd: RDD[RawPanda]) = {
    rdd.map(p => (p.zip, (p.attributes(0), 1))).reduceByKey{case (x, y) => (x._1 + y._1, x._2 + y._2)}.collect()
  }<commit_msg>Add grouped timings<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.highperformancespark.examples.perf

import com.highperformancespark.examples.dataframe.RawPanda
import com.highperformancespark.examples.tools._

import org.apache.spark.rdd._
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.{DataFrame, Dataset, Row}
import org.apache.spark.sql.hive.HiveContext

/**
 * A simple performance test to compare a simple sort between DataFrame, and RDD
 */
object SimplePerfTest {
  def main(args: Array[String]) = {
    val sparkConf = new SparkConf().setAppName("simple-perf-test")
    val sc = new SparkContext(sparkConf)
    val sqlCtx = new HiveContext(sc)
    val scalingFactor = if (args.length > 0) args(0).toLong else 100L
    val size = if (args.length > 1) args(1).toInt else 50
    run(sc, sqlCtx, scalingFactor, size)
  }

  def run(sc: SparkContext, sqlCtx: HiveContext, scalingFactor: Long, size: Int) = {
    import sqlCtx.implicits._
    val inputRDD = GenerateScalingData.generateFullGoldilocks(sc, scalingFactor, size)
    inputRDD.cache()
    inputRDD.count()
    val rddTimeings = 1.to(10).map(x => time(testOnRDD(inputRDD)))
    val groupTimeings = 1.to(10).map(x => time(groupOnRDD(inputRDD)))
    val inputDataFrame = inputRDD.toDF()
    inputDataFrame.cache()
    inputDataFrame.count()
    val dataFrameTimeings = 1.to(10).map(x => time(testOnDataFrame(inputDataFrame)))
    println(rddTimeings.map(_._2).mkString(","))
    println(groupTimeings.map(_._2).mkString(","))
    println(dataFrameTimeings.map(_._2).mkString(","))
  }

  def groupOnRDD(rdd: RDD[RawPanda]) = {
    rdd.map(p => (p.zip, (p.attributes(0), 1))).reduceByKey{case (x, y) => (x._1 + y._1, x._2 + y._2)}.collect()
  }

  def testOnRDD(rdd: RDD[RawPanda]) = {
    rdd.map(p => (p.zip, p.attributes(0))).groupByKey().mapValues{v =>
      v.aggregate((0.0, 0))({case (x, y) => (x._1 + y, x._2 + 1)},
        {case (x, y) => (x._1 + y._1, x._2 + y._2)})}
  }
{'fraction_non_alphanumeric': 0.12108843537414966, 'fraction_numerical': 0.013411078717201166, 'mean_word_length': 4.090009891196835, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '142121', 'n_tokens_mistral': 1684, 'n_tokens_neox': 1626, 'n_words': 480}