要求对统计的单词的次数降序排列：
java版本：
private static final long serialVersionUID = 1L;
private static final long serialVersionUID = 1L;
private static final long serialVersionUID = 1L;
private static final long serialVersionUID = 1L;
private static final long serialVersionUID = 1L;
scala版本：
java版本：
package cn.spark.study.core;
import java.util.Arrays;
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.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
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.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
public class wordcountsort {
public static void main(String[] args) {
SparkConf conf = new SparkConf()
.setAppName("WorldCountLocal")
.setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD<String> lines = sc.textFile("C://Users//hlz//Desktop/test.txt");
JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>(){
public static void main(String[] args) {
SparkConf conf = new SparkConf()
.setAppName("WorldCountLocal")
.setMaster("local");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD<String> lines = sc.textFile("C://Users//hlz//Desktop/test.txt");
JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>(){
private static final long serialVersionUID = 1L;
@Override
public Iterable<String> call(String line) throws Exception {
// TODO Auto-generated method stub
return Arrays.asList(line.split(" "));
}
});
JavaPairRDD<String, Integer> pairs = words.mapToPair(
new PairFunction<String, String, Integer>(){
public Iterable<String> call(String line) throws Exception {
// TODO Auto-generated method stub
return Arrays.asList(line.split(" "));
}
});
JavaPairRDD<String, Integer> pairs = words.mapToPair(
new PairFunction<String, String, Integer>(){
private static final long serialVersionUID = 1L;
@Override
public Tuple2<String, Integer> call(String word) throws Exception {
// TODO Auto-generated method stub
return new Tuple2<String, Integer>(word,1);
}
});
JavaPairRDD<String, Integer> wordCounts = pairs.reduceByKey(
new Function2<Integer, Integer, Integer>(){
public Tuple2<String, Integer> call(String word) throws Exception {
// TODO Auto-generated method stub
return new Tuple2<String, Integer>(word,1);
}
});
JavaPairRDD<String, Integer> wordCounts = pairs.reduceByKey(
new Function2<Integer, Integer, Integer>(){
private static final long serialVersionUID = 1L;
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
// TODO Auto-generated method stub
return v1 + v2;
}
});
JavaPairRDD<Integer, String> rewordcount = wordCounts.mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>(){
public Integer call(Integer v1, Integer v2) throws Exception {
// TODO Auto-generated method stub
return v1 + v2;
}
});
JavaPairRDD<Integer, String> rewordcount = wordCounts.mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>(){
private static final long serialVersionUID = 1L;
@Override
public Tuple2<Integer, String> call(Tuple2<String, Integer> t) throws Exception {
return new Tuple2<Integer,String>(t._2,t._1);
}
});
JavaPairRDD<Integer, String> sortwordcount = rewordcount.sortByKey(false);
JavaPairRDD<String, Integer> finalwordcount = sortwordcount.mapToPair(new PairFunction<Tuple2<Integer,String>, String,Integer >(){
public Tuple2<Integer, String> call(Tuple2<String, Integer> t) throws Exception {
return new Tuple2<Integer,String>(t._2,t._1);
}
});
JavaPairRDD<Integer, String> sortwordcount = rewordcount.sortByKey(false);
JavaPairRDD<String, Integer> finalwordcount = sortwordcount.mapToPair(new PairFunction<Tuple2<Integer,String>, String,Integer >(){
private static final long serialVersionUID = 1L;
@Override
public Tuple2<String, Integer> call(Tuple2<Integer, String> t) throws Exception {
// TODO Auto-generated method stub
return new Tuple2<String,Integer>(t._2,t._1);
}
});
finalwordcount.foreach(new VoidFunction<Tuple2<String, Integer>>(){
public Tuple2<String, Integer> call(Tuple2<Integer, String> t) throws Exception {
// TODO Auto-generated method stub
return new Tuple2<String,Integer>(t._2,t._1);
}
});
finalwordcount.foreach(new VoidFunction<Tuple2<String, Integer>>(){
private static final long serialVersionUID = 1L;
@Override
public void call(Tuple2<String, Integer> wordCount) throws Exception {
System.out.println(wordCount._1 + " append " + wordCount._2 + " times. ");
}
});
sc.close();
}
public void call(Tuple2<String, Integer> wordCount) throws Exception {
System.out.println(wordCount._1 + " append " + wordCount._2 + " times. ");
}
});
sc.close();
}
}
scala版本：
package com.spark.study.core
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext
object wordcountsort {
def main(args:Array[String]){
val conf = new SparkConf()
.setAppName("teshurddcaozuo")
.setMaster("local");
val sc = new SparkContext(conf);
val lines = sc.textFile("C://Users//hlz//Desktop/test.txt", 1)
val linepair = lines.flatMap(line => line.split(" "))
val cccount = linepair.map(f=>(f,1))
val linecount = cccount.reduceByKey(_ + _)
val revecount = linecount.map(line =>(line._2,line._1))
val sortcount = revecount.sortByKey(false)
val recount = sortcount.map(line =>(line._2,line._1))
recount.foreach(line =>println(line._1 + " appear " + line._2 + " times "))
}
}
def main(args:Array[String]){
val conf = new SparkConf()
.setAppName("teshurddcaozuo")
.setMaster("local");
val sc = new SparkContext(conf);
val lines = sc.textFile("C://Users//hlz//Desktop/test.txt", 1)
val linepair = lines.flatMap(line => line.split(" "))
val cccount = linepair.map(f=>(f,1))
val linecount = cccount.reduceByKey(_ + _)
val revecount = linecount.map(line =>(line._2,line._1))
val sortcount = revecount.sortByKey(false)
val recount = sortcount.map(line =>(line._2,line._1))
recount.foreach(line =>println(line._1 + " appear " + line._2 + " times "))
}
}
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/30541278/viewspace-2153550/ ，如需转载，请注明出处，否则将追究法律责任。
请登录后发表评论
登录
全部评论
最新文章
- spark系列二：jdbc数据源实例
- spark系列一：共享变量(broadcast和accumulator)
- python实例方法、类方法和静态方法的区别
- spark系列二：sparkcore和sparksql综合案例
- hadoop启动时，报ssh: Could not resolve hostname xxx: Name or service not known
- spark系列一：使用持久化机制和不使用持久化的原理及实战
- spark系列一：wordcount程序原理剖析及spark架构原理
- spark系列一：transaction和action介绍
- spark系列二：sparksql dataframe常用操作
- spark系列三：sparkstreaming之输入DStream之Kafka数据源实战（基于Direct的方式）
