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實現(xiàn)單詞計數(shù)

??*/

object SparkWordCount {

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

????/**

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

??????*/

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

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

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

?

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

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

?

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

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

?

????// 切分數(shù)據(jù)，生成一個個單詞

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

?

????// 把單詞生成一個個元組

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

?

????// 進行聚合操作

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

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

?

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

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

?

????// 打印到控制臺

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

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

// ???sorted.foreach(println)

?

????// 把結(jié)果存儲到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<String> lines = jsc.textFile("hdfs://hadoop01:9000/wordcount/input/a.txt");

????????//切分數(shù)據(jù)

????????JavaRDD<String> words =

lines.flatMap(new FlatMapFunction<String, String>() {

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

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

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

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

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

????????});

?

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

????????JavaPairRDD<String, Integer> tuples =

words.mapToPair(new PairFunction<String, String, Integer>() {

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

????????????public Tuple2<String, Integer> call(String s) throws Exception {

????????????????return new Tuple2<String, Integer>(s, 1);

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

????????});

?

????????//聚合 ?????????????????????????????????????????????????//2個相同key的value,聚合

????????JavaPairRDD<String, Integer> sumed =

tuples.reduceByKey(new Function2<Integer, Integer, Integer>() {

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

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

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

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

????????});

?

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

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

????????final JavaPairRDD<Integer, String> swaped =

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

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

????????????public Tuple2<Integer, String> call(Tuple2<String, Integer> tup) throws Exception {

// ???????????????return new Tuple2<Integer, String>(tup._2, tup._1);

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

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

????????});

?

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

????????JavaPairRDD<Integer, String> sorted = swaped.sortByKey(false);

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

????????JavaPairRDD<String, Integer> res = sorted.mapToPair(

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

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

???????????????public Tuple2<String, Integer> call(Tuple2<Integer, String> tup)throws Exception {

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

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

????????});

?

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

?

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

????}

}