大数据学习路线分享Master的jps

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实现单词计数

*/

object SparkWordCount {

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

/**

* 构建模板代码

*/

val conf: SparkConf = new SparkConf()

.setAppName("SparkWordCount")

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

// 创建提交任务到集群的入口类（上下文对象）

val sc: SparkContext = new SparkContext(conf)

// 获取HDFS的数据

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

// 切分数据，生成一个个单词

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(_+_)

// 以单词出现的次数进行降序排序

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

// 打印到控制台

// println(sorted.collect.toBuffer)

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

// sorted.foreach(println)

// 把结果存储到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]");

//提交任务入口类

JavaSparkContext jsc = new JavaSparkContext(conf);

//获取数据

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

//切分数据

JavaRDD words =

lines.flatMap(new FlatMapFunction() {

@Override

public Iterable call(String s) throws Exception {

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

return splited;

}

});

//生成元祖 //一对一组 ,(输入单词,输出单词,输出1)

JavaPairRDD tuples =

words.mapToPair(new PairFunction() {

@Override

public Tuple2 call(String s) throws Exception {

return new Tuple2(s, 1);

}

});

//聚合 //2个相同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算子，此时需要把两个值位置调换，排序完成后，在换回来

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();//释放资源

}

}