Merge pull request minyk#1 from minyk/master

Fix some issues.
nexr · May 5, 2015 · 716a791 · 716a791
2 parents aa31fff + a020f3d
commit 716a791
Show file tree

Hide file tree

Showing 3 changed files with 2 additions and 6 deletions.
diff --git a/src/main/resources/run.sh b/src/main/resources/run.sh
@@ -251,7 +251,7 @@ if (($benchmark_result == 1))
 then
 total_time=`expr ${end} - ${start}`
 total_time_in_hour=$(echo "scale=4;${total_time}/3600" | bc)
-scale_factor=$(echo "scale=4;${hssize}/10000000000" | bc)
+scale_factor=$(echo "scale=4;${hssize}/1000000000000" | bc)
 perf_metric=$(echo "scale=4;${scale_factor}/${total_time_in_hour}" | bc)
 
 echo -e "${green}$sep============${NC}" | tee -a ./result-"${prefix}".log

diff --git a/src/main/scala/com/nexr/spark/terasort/TeraInputFormat.scala b/src/main/scala/com/nexr/spark/terasort/TeraInputFormat.scala
@@ -34,10 +34,6 @@ object TeraInputFormat {
   val KEY_LEN = 10
   val VALUE_LEN = 90
   val RECORD_LEN = KEY_LEN + VALUE_LEN
-  val NUM_INPUT_FILES = "mapreduce.input.fileinputformat.numinputfiles"
-  val NUM_PARTITIONS = "mapreduce.terasort.num.partitions"
-  val SAMPLE_SIZE = "mapreduce.terasort.partitions.sample"
-  val SPLIT_SLOP = 1.1
   var lastContext : JobContext = null
   var lastResult : List[InputSplit] = null
   implicit val caseInsensitiveOrdering = UnsignedBytes.lexicographicalComparator

diff --git a/src/main/scala/com/nexr/spark/terasort/TeraSort.scala b/src/main/scala/com/nexr/spark/terasort/TeraSort.scala
@@ -38,7 +38,7 @@ object TeraSort {
     val sc = new SparkContext(conf)
 
     val dataset = sc.newAPIHadoopFile[Array[Byte], Array[Byte], TeraInputFormat](inputFile)
-    val sorted = dataset.sortByKey()
+    val sorted = dataset.partitionBy(new TeraSortPartitioner(dataset.partitions.length)).sortByKey()
     sorted.saveAsNewAPIHadoopFile[TeraOutputFormat](outputFile)
   }
 }