scala> 1.to(10)

res0: scala.collection.immutable.Range.Inclusive = Range(1, 2, 3, 4, 5, 6, 7, 8,

?9, 10)

?

scala> 1 to 10

res1: scala.collection.immutable.Range.Inclusive = Range(1, 2, 3, 4, 5, 6, 7, 8,

?9, 10)

/**

??* 用spark實(shí)現(xiàn)單詞計(jì)數(shù)

??*/

object SparkWordCount {

??def main(args: Array[String]): Unit = {

????/**

??????* 構(gòu)建模板代碼

??????*/

????val conf: SparkConf = new SparkConf()

??????.setAppName("SparkWordCount")

// ?????.setMaster("local[2]")

?

????// 創(chuàng)建提交任務(wù)到集群的入口類（上下文對(duì)象）

????val sc: SparkContext = new SparkContext(conf)

?

????// 獲取HDFS的數(shù)據(jù)

????val lines: RDD[String] = sc.textFile(args(0))

?

????// 切分?jǐn)?shù)據(jù)，生成一個(gè)個(gè)單詞

????val words: RDD[String] = lines.flatMap(_.split(" "))

?

????// 把單詞生成一個(gè)個(gè)元組

????val tuples: RDD[(String, Int)] = words.map((_, 1))

?

????// 進(jìn)行聚合操作

// ???tuples.reduceByKey((x, y) => x + y)

????val sumed: RDD[(String, Int)] = tuples.reduceByKey(_+_)

?

????// 以單詞出現(xiàn)的次數(shù)進(jìn)行降序排序

????val sorted: RDD[(String, Int)] = sumed.sortBy(_._2, false)

?

????// 打印到控制臺(tái)

// ???println(sorted.collect.toBuffer)

// ???sorted.foreach(x => println(x))

// ???sorted.foreach(println)

?

????// 把結(jié)果存儲(chǔ)到HDFS

????sorted.saveAsTextFile(args(1))

?

????// 釋放資源

????sc.stop()

??}

}

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

?

import java.util.Arrays;

import java.util.List;

?

public class JavaSparkWC {

????public static void main(String[] args) {

????????SparkConf conf = new SparkConf()

????????????????.setAppName("JavaSparkWC").setMaster("local[1]");

????????

//提交任務(wù)入口類

????????JavaSparkContext jsc = new JavaSparkContext(conf);

?

????????//獲取數(shù)據(jù)

????????JavaRDD lines = jsc.textFile("hdfs://hadoop01:9000/wordcount/input/a.txt");

????????//切分?jǐn)?shù)據(jù)

????????JavaRDD words =

lines.flatMap(new FlatMapFunction() {

????????????@Override

????????????public Iterable call(String s) throws Exception {

????????????????List splited = Arrays.asList(s.split(" ")); //生成list

????????????????return splited;

????????????}

????????});

?

????????//生成元祖 ??????????????????????????????//一對(duì)一組 ,(輸入單詞,輸出單詞,輸出1)

????????JavaPairRDD tuples =

words.mapToPair(new PairFunction() {

????????????@Override

????????????public Tuple2 call(String s) throws Exception {

????????????????return new Tuple2(s, 1);

????????????}

????????});

?

????????//聚合 ?????????????????????????????????????????????????//2個(gè)相同key的value,聚合

????????JavaPairRDD sumed =

tuples.reduceByKey(new Function2() {

????????????@Override

????????????public Integer call(Integer v1, Integer v2) throws Exception {

????????????????return v1 + v2;

????????????}

????????});

?

????????//此前key為String類型,沒有辦法排序

????????//Java api并沒有提供sortBy算子，此時(shí)需要把兩個(gè)值位置調(diào)換，排序完成后，在換回來

????????final JavaPairRDD swaped =

sumed.mapToPair(new PairFunction, Integer, String>() {

????????????@Override

????????????public Tuple2 call(Tuple2 tup) throws Exception {

// ???????????????return new Tuple2(tup._2, tup._1);

????????????????return tup.swap(); //swap(),交換方法

????????????}

????????});

?

????????//降序排序

????????JavaPairRDD sorted = swaped.sortByKey(false);

????????//再次交換

????????JavaPairRDD res = sorted.mapToPair(

????????????new PairFunction, String, Integer>() {

???????????????@Override

???????????????public Tuple2 call(Tuple2 tup)throws Exception {

????????????????????return tup.swap();

???????????????}

????????});

?

????????System.out.println(res.collect());

?

????????jsc.stop();//釋放資源

????}

}