Hadoop vs Spark性能对比 ...

然后对RDD进行map操作，抽取每个vector（point）对应的类号，输出（K,V）为（class，（Point，1）），组成新的RDD。

然后再reduce之前，对每个新的RDD进行combine，在RDD内部算出每个class的中心和。使得每个RDD的输出只有最多K个KV对。

最后进行reduce得到新的RDD（内容的Key是class，Value是中心和，再经过map后得到最后的中心。

Final centers: Map(5 -> (13.997101228817169, 9.208875044622895, -2.494072457488311), 8 -> (-2.33522333047955, 9.128892414676326, 1.7923150585737604), 7 -> (8.658031587043952, 2.162306996983008, 17.670646829079146), 3 -> (11.530154433698268, 0.17834347219956842, 9.224352885937776), 4 -> (12.722903153986868, 8.812883284216143, 0.6564509961064319), 1 -> (6.458644369071984, 11.345681702383024, 7.041924994173552), 6 -> (12.887793408866614, -1.5189406469928937, 9.526393664105957), 2 -> (2.3345459304412164, 2.0173098597285533, 1.4772489989976143))

0/user/LijieXu/Kmeans/Square-20GB.txt 8 0.8

import spark._

object HdfsTest {

def main(args: Array[String]) {

val sc = new SparkContext(args(0), “HdfsTest”)

val file = sc.textFile(args(1))

val mapped = file.map(s => s.length).cache()

for (iter <- 1 to 10) {

val start = System.currentTimeMillis()

for (x <- mapped) { x + 2 }

// println(“Processing: ” + x)

val end = System.currentTimeMillis()

println(“Iteration ” + iter + ” took ” + (end-start) + ” ms”)

}

}

}

Iteration 2 took 388 ms

Iteration 3 took 472 ms

Iteration 4 took 490 ms

Iteration 5 took 459 ms

Iteration 6 took 492 ms

Iteration 7 took 480 ms

Iteration 8 took 501 ms

Iteration 9 took 479 ms

Iteration 10 took 432 ms

import SparkContext._

object WordCount {

def main(args: Array[String]) {

if (args.length < 2) {

System.err.println(“Usage: wordcount <master> <jar>”)

System.exit(1)

}

val sp = new SparkContext(args(0), “wordcount”, “/opt/spark”, List(args(1)))

val file = sp.textFile(“hdfs://master:9000/user/LijieXu/Wikipedia/txt/enwiki-20110405.txt”);

val counts = file.flatMap(line => line.split(” “)).map(word => (word, 1)).reduceByKey(_ + _)

counts.saveAsTextFile(“hdfs://master:9000/user/Output/WikiResult3″)

}

}

scala> file.first

scala> val words = file.map(_.split(‘ ‘)).filter(_.size < 100) //得到RDD[Array[String]]

scala> words.cache

scala> words.filter(_.contains(“Beijing”)).count

12/06/06 22:12:33 INFO SparkContext: Job finished in 10.862765819 s

res1: Long = 855

scala> words.filter(_.contains(“Beijing”)).count

12/06/06 22:12:52 INFO SparkContext: Job finished in 0.71051464 s

res2: Long = 855

scala> words.filter(_.contains(“Shanghai”)).count

12/06/06 22:13:23 INFO SparkContext: Job finished in 0.667734427 s

res3: Long = 614

scala> words.filter(_.contains(“Guangzhou”)).count

12/06/06 22:13:42 INFO SparkContext: Job finished in 0.800617719 s

res4: Long = 134