- map()
- filter()
- flatMap()
- groupByKey()
- reduceByKey()
- sortByKey()
- join()
- cogroup()
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
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.VoidFunction;
import scala.Tuple2;
/**
* transformation操作實(shí)戰(zhàn)
* @author Administrator
*
*/
@SuppressWarnings(value = {"unused", "unchecked"})
public class TransformationOperation {
public static void main(String[] args) {
// map();
// filter();
// flatMap();
// groupByKey();
// reduceByKey();
// sortByKey();
// join();
cogroup();
}
/**
* map算子案例：將集合中每一個(gè)元素都乘以2
*/
private static void map() {
// 創(chuàng)建SparkConf
SparkConf conf = new SparkConf()
.setAppName("map")
.setMaster("local");
// 創(chuàng)建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 構(gòu)造集合
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
// 并行化集合享怀，創(chuàng)建初始RDD
JavaRDD<Integer> numberRDD = sc.parallelize(numbers);
// 使用map算子，將集合中的每個(gè)元素都乘以2
// map算子趟咆，是對(duì)任何類型的RDD添瓷，都可以調(diào)用的
// 在java中，map算子接收的參數(shù)是Function對(duì)象
// 創(chuàng)建的Function對(duì)象值纱，一定會(huì)讓你設(shè)置第二個(gè)泛型參數(shù)鳞贷，這個(gè)泛型類型，就是返回的新元素的類型
// 同時(shí)call()方法的返回類型虐唠，也必須與第二個(gè)泛型類型同步
// 在call()方法內(nèi)部搀愧，就可以對(duì)原始RDD中的每一個(gè)元素進(jìn)行各種處理和計(jì)算眷蚓，并返回一個(gè)新的元素
// 所有新的元素就會(huì)組成一個(gè)新的RDD
JavaRDD<Integer> multipleNumberRDD = numberRDD.map(
new Function<Integer, Integer>() {
private static final long serialVersionUID = 1L;
// 傳入call()方法的枫疆，就是1,2,3,4,5
// 返回的就是2,4,6,8,10
@Override
public Integer call(Integer v1) throws Exception {
return v1 * 2;
}
});
// 打印新的RDD
multipleNumberRDD.foreach(new VoidFunction<Integer>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Integer t) throws Exception {
System.out.println(t);
}
});
// 關(guān)閉JavaSparkContext
sc.close();
}
/**
* filter算子案例：過濾集合中的偶數(shù)
*/
private static void filter() {
// 創(chuàng)建SparkConf
SparkConf conf = new SparkConf()
.setAppName("filter")
.setMaster("local");
// 創(chuàng)建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 模擬集合
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
// 并行化集合愿险，創(chuàng)建初始RDD
JavaRDD<Integer> numberRDD = sc.parallelize(numbers);
// 對(duì)初始RDD執(zhí)行filter算子疑苔，過濾出其中的偶數(shù)
// filter算子刻恭，傳入的也是Function，其他的使用注意點(diǎn)，實(shí)際上和map是一樣的
// 但是呛凶，唯一的不同围橡，就是call()方法的返回類型是Boolean
// 每一個(gè)初始RDD中的元素塞赂，都會(huì)傳入call()方法，此時(shí)你可以執(zhí)行各種自定義的計(jì)算邏輯
// 來判斷這個(gè)元素是否是你想要的
// 如果你想在新的RDD中保留這個(gè)元素延欠，那么就返回true为居；否則恭陡，不想保留這個(gè)元素，返回false
JavaRDD<Integer> evenNumberRDD = numberRDD.filter(
new Function<Integer, Boolean>() {
private static final long serialVersionUID = 1L;
// 在這里苔埋，1到10荞膘，都會(huì)傳入進(jìn)來
// 但是根據(jù)我們的邏輯费奸，只有2,4,6,8,10這幾個(gè)偶數(shù)，會(huì)返回true
// 所以习劫，只有偶數(shù)會(huì)保留下來，放在新的RDD中
@Override
public Boolean call(Integer v1) throws Exception {
return v1 % 2 == 0; // 注意返回的是Boolean
}
});
// 打印新的RDD
evenNumberRDD.foreach(new VoidFunction<Integer>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Integer t) throws Exception {
System.out.println(t);
}
});
// 關(guān)閉JavaSparkContext
sc.close();
}
/**
* flatMap案例：將文本行拆分為多個(gè)單詞
*/
private static void flatMap() {
// 創(chuàng)建SparkConf
SparkConf conf = new SparkConf()
.setAppName("flatMap")
.setMaster("local");
// 創(chuàng)建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 構(gòu)造集合
List<String> lineList = Arrays.asList("hello you", "hello me", "hello world");
// 并行化集合泄隔，創(chuàng)建RDD
JavaRDD<String> lines = sc.parallelize(lineList);
// 對(duì)RDD執(zhí)行flatMap算子拒贱，將每一行文本，拆分為多個(gè)單詞
// flatMap算子佛嬉，在java中逻澳，接收的參數(shù)是FlatMapFunction
// 我們需要自己定義FlatMapFunction的第二個(gè)泛型類型，即暖呕，代表了返回的新元素的類型
// call()方法斜做，返回的類型，不是U湾揽，而是Iterable<U>瓤逼，這里的U也與第二個(gè)泛型類型相同
// flatMap其實(shí)就是，接收原始RDD中的每個(gè)元素库物，并進(jìn)行各種邏輯的計(jì)算和處理霸旗，返回可以返回多個(gè)元素
// 多個(gè)元素，即封裝在Iterable集合中戚揭，可以使用ArrayList等集合
// 新的RDD中诱告，即封裝了所有的新元素；也就是說民晒，新的RDD的大小一定是 >= 原始RDD的大小
JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
private static final long serialVersionUID = 1L;
// 在這里會(huì)精居，比如，傳入第一行潜必，hello you
// 返回的是一個(gè)Iterable<String>(hello, you)
@Override
public Iterable<String> call(String t) throws Exception {
return Arrays.asList(t.split(" "));
}
});
// 打印新的RDD
words.foreach(new VoidFunction<String>() {
private static final long serialVersionUID = 1L;
@Override
public void call(String t) throws Exception {
System.out.println(t);
}
});
// 關(guān)閉JavaSparkContext
sc.close();
}
/**
* groupByKey案例：按照班級(jí)對(duì)成績(jī)進(jìn)行分組
*/
private static void groupByKey() {
// 創(chuàng)建SparkConf
SparkConf conf = new SparkConf()
.setAppName("groupByKey")
.setMaster("local");
// 創(chuàng)建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 模擬集合
List<Tuple2<String, Integer>> scoreList = Arrays.asList(
new Tuple2<String, Integer>("class1", 80),
new Tuple2<String, Integer>("class2", 75),
new Tuple2<String, Integer>("class1", 90),
new Tuple2<String, Integer>("class2", 65));
// 并行化集合靴姿，創(chuàng)建JavaPairRDD
JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);
// 針對(duì)scores RDD，執(zhí)行g(shù)roupByKey算子磁滚，對(duì)每個(gè)班級(jí)的成績(jī)進(jìn)行分組
// groupByKey算子空猜，返回的還是JavaPairRDD
// 但是，JavaPairRDD的第一個(gè)泛型類型不變恨旱，第二個(gè)泛型類型變成Iterable這種集合類型
// 也就是說辈毯，按照了key進(jìn)行分組，那么每個(gè)key可能都會(huì)有多個(gè)value搜贤，此時(shí)多個(gè)value聚合成了Iterable
// 那么接下來谆沃，我們是不是就可以通過groupedScores這種JavaPairRDD，很方便地處理某個(gè)分組內(nèi)的數(shù)據(jù)
JavaPairRDD<String, Iterable<Integer>> groupedScores = scores.groupByKey();
// 打印groupedScores RDD
groupedScores.foreach(new VoidFunction<Tuple2<String,Iterable<Integer>>>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Tuple2<String, Iterable<Integer>> t)
throws Exception {
System.out.println("class: " + t._1);
Iterator<Integer> ite = t._2.iterator();
while(ite.hasNext()) {
System.out.println(ite.next());
}
System.out.println("==============================");
}
});
// 關(guān)閉JavaSparkContext
sc.close();
}
/**
* reduceByKey案例：統(tǒng)計(jì)每個(gè)班級(jí)的總分
*/
private static void reduceByKey() {
// 創(chuàng)建SparkConf
SparkConf conf = new SparkConf()
.setAppName("reduceByKey")
.setMaster("local");
// 創(chuàng)建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 模擬集合
List<Tuple2<String, Integer>> scoreList = Arrays.asList(
new Tuple2<String, Integer>("class1", 80),
new Tuple2<String, Integer>("class2", 75),
new Tuple2<String, Integer>("class1", 90),
new Tuple2<String, Integer>("class2", 65));
// 并行化集合仪芒，創(chuàng)建JavaPairRDD
JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);
// 針對(duì)scores RDD唁影，執(zhí)行reduceByKey算子
// reduceByKey耕陷，接收的參數(shù)是Function2類型，它有三個(gè)泛型參數(shù)据沈，實(shí)際上代表了三個(gè)值
// 第一個(gè)泛型類型和第二個(gè)泛型類型哟沫，代表了原始RDD中的元素的value的類型
// 因此對(duì)每個(gè)key進(jìn)行reduce，都會(huì)依次將第一個(gè)锌介、第二個(gè)value傳入嗜诀，將值再與第三個(gè)value傳入
// 因此此處，會(huì)自動(dòng)定義兩個(gè)泛型類型孔祸，代表call()方法的兩個(gè)傳入?yún)?shù)的類型
// 第三個(gè)泛型類型隆敢，代表了每次reduce操作返回的值的類型，默認(rèn)也是與原始RDD的value類型相同的
// reduceByKey算法返回的RDD崔慧，還是JavaPairRDD<key, value>
JavaPairRDD<String, Integer> totalScores = scores.reduceByKey(
new Function2<Integer, Integer, Integer>() {
private static final long serialVersionUID = 1L;
// 對(duì)每個(gè)key拂蝎，都會(huì)將其value，依次傳入call方法
// 從而聚合出每個(gè)key對(duì)應(yīng)的一個(gè)value
// 然后惶室，將每個(gè)key對(duì)應(yīng)的一個(gè)value温自，組合成一個(gè)Tuple2，作為新RDD的元素
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
// 打印totalScores RDD
totalScores.foreach(new VoidFunction<Tuple2<String,Integer>>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Tuple2<String, Integer> t) throws Exception {
System.out.println(t._1 + ": " + t._2);
}
});
// 關(guān)閉JavaSparkContext
sc.close();
}
/**
* sortByKey案例：按照學(xué)生分?jǐn)?shù)進(jìn)行排序
*/
private static void sortByKey() {
// 創(chuàng)建SparkConf
SparkConf conf = new SparkConf()
.setAppName("sortByKey")
.setMaster("local");
// 創(chuàng)建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 模擬集合
List<Tuple2<Integer, String>> scoreList = Arrays.asList(
new Tuple2<Integer, String>(65, "leo"),
new Tuple2<Integer, String>(50, "tom"),
new Tuple2<Integer, String>(100, "marry"),
new Tuple2<Integer, String>(80, "jack"));
// 并行化集合皇钞，創(chuàng)建RDD
JavaPairRDD<Integer, String> scores = sc.parallelizePairs(scoreList);
// 對(duì)scores RDD執(zhí)行sortByKey算子
// sortByKey其實(shí)就是根據(jù)key進(jìn)行排序捣作，可以手動(dòng)指定升序，或者降序
// 返回的鹅士，還是JavaPairRDD券躁，其中的元素內(nèi)容，都是和原始的RDD一模一樣的
// 但是就是RDD中的元素的順序掉盅，不同了
JavaPairRDD<Integer, String> sortedScores = scores.sortByKey(false);
// 打印sortedScored RDD
sortedScores.foreach(new VoidFunction<Tuple2<Integer,String>>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Tuple2<Integer, String> t) throws Exception {
System.out.println(t._1 + ": " + t._2);
}
});
// 關(guān)閉JavaSparkContext
sc.close();
}
/**
* join案例：打印學(xué)生成績(jī)
*/
private static void join() {
// 創(chuàng)建SparkConf
SparkConf conf = new SparkConf()
.setAppName("join")
.setMaster("local");
// 創(chuàng)建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 模擬集合
List<Tuple2<Integer, String>> studentList = Arrays.asList(
new Tuple2<Integer, String>(1, "leo"),
new Tuple2<Integer, String>(2, "jack"),
new Tuple2<Integer, String>(3, "tom"));
List<Tuple2<Integer, Integer>> scoreList = Arrays.asList(
new Tuple2<Integer, Integer>(1, 100),
new Tuple2<Integer, Integer>(2, 90),
new Tuple2<Integer, Integer>(3, 60));
// 并行化兩個(gè)RDD
JavaPairRDD<Integer, String> students = sc.parallelizePairs(studentList);
JavaPairRDD<Integer, Integer> scores = sc.parallelizePairs(scoreList);
// 使用join算子關(guān)聯(lián)兩個(gè)RDD
// join以后也拜，還是會(huì)根據(jù)key進(jìn)行join，并返回JavaPairRDD
// 但是JavaPairRDD的第一個(gè)泛型類型趾痘，之前兩個(gè)JavaPairRDD的key的類型慢哈，因?yàn)槭峭ㄟ^key進(jìn)行join的
// 第二個(gè)泛型類型，是Tuple2<v1, v2>的類型永票，Tuple2的兩個(gè)泛型分別為原始RDD的value的類型
// join卵贱，就返回的RDD的每一個(gè)元素，就是通過key join上的一個(gè)pair
// 什么意思呢侣集？比如有(1, 1) (1, 2) (1, 3)的一個(gè)RDD
// 還有一個(gè)(1, 4) (2, 1) (2, 2)的一個(gè)RDD
// join以后键俱，實(shí)際上會(huì)得到(1 (1, 4)) (1, (2, 4)) (1, (3, 4))
JavaPairRDD<Integer, Tuple2<String, Integer>> studentScores = students.join(scores);
// 打印studnetScores RDD
studentScores.foreach(
new VoidFunction<Tuple2<Integer,Tuple2<String,Integer>>>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Tuple2<Integer, Tuple2<String, Integer>> t)
throws Exception {
System.out.println("student id: " + t._1);
System.out.println("student name: " + t._2._1);
System.out.println("student score: " + t._2._2);
System.out.println("===============================");
}
});
// 關(guān)閉JavaSparkContext
sc.close();
}
/**
* cogroup案例：打印學(xué)生成績(jī)
*/
private static void cogroup() {
// 創(chuàng)建SparkConf
SparkConf conf = new SparkConf()
.setAppName("cogroup")
.setMaster("local");
// 創(chuàng)建JavaSparkContext
JavaSparkContext sc = new JavaSparkContext(conf);
// 模擬集合
List<Tuple2<Integer, String>> studentList = Arrays.asList(
new Tuple2<Integer, String>(1, "leo"),
new Tuple2<Integer, String>(2, "jack"),
new Tuple2<Integer, String>(3, "tom"));
List<Tuple2<Integer, Integer>> scoreList = Arrays.asList(
new Tuple2<Integer, Integer>(1, 100),
new Tuple2<Integer, Integer>(2, 90),
new Tuple2<Integer, Integer>(3, 60),
new Tuple2<Integer, Integer>(1, 70),
new Tuple2<Integer, Integer>(2, 80),
new Tuple2<Integer, Integer>(3, 50));
// 并行化兩個(gè)RDD
JavaPairRDD<Integer, String> students = sc.parallelizePairs(studentList);
JavaPairRDD<Integer, Integer> scores = sc.parallelizePairs(scoreList);
// cogroup與join不同
// 相當(dāng)于是，一個(gè)key join上的所有value世分，都給放到一個(gè)Iterable里面去了
// cogroup编振，不太好講解，希望大家通過動(dòng)手編寫我們的案例臭埋，仔細(xì)體會(huì)其中的奧妙
JavaPairRDD<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> studentScores =
students.cogroup(scores);
// 打印studnetScores RDD
studentScores.foreach(
new VoidFunction<Tuple2<Integer,Tuple2<Iterable<String>,Iterable<Integer>>>>() {
private static final long serialVersionUID = 1L;
@Override
public void call(
Tuple2<Integer, Tuple2<Iterable<String>, Iterable<Integer>>> t)
throws Exception {
System.out.println("student id: " + t._1);
System.out.println("student name: " + t._2._1);
System.out.println("student score: " + t._2._2);
// student id: 1
// student name: [leo]
// student score: [100, 70]
System.out.println("===============================");
}
});
// 關(guān)閉JavaSparkContext
sc.close();
}
}
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
/**
* @author Administrator
*/
object TransformationOperation {
def main(args: Array[String]) {
// map()
// filter()
// flatMap()
// groupByKey()
// reduceByKey()
// sortByKey()
join()
}
def map() {
val conf = new SparkConf()
.setAppName("map")
.setMaster("local")
val sc = new SparkContext(conf)
val numbers = Array(1, 2, 3, 4, 5)
val numberRDD = sc.parallelize(numbers, 1)
val multipleNumberRDD = numberRDD.map { num => num * 2 }
multipleNumberRDD.foreach { num => println(num) }
}
def filter() {
val conf = new SparkConf()
.setAppName("filter")
.setMaster("local")
val sc = new SparkContext(conf)
val numbers = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val numberRDD = sc.parallelize(numbers, 1)
val evenNumberRDD = numberRDD.filter { num => num % 2 == 0 }
evenNumberRDD.foreach { num => println(num) }
}
def flatMap() {
val conf = new SparkConf()
.setAppName("flatMap")
.setMaster("local")
val sc = new SparkContext(conf)
val lineArray = Array("hello you", "hello me", "hello world")
val lines = sc.parallelize(lineArray, 1)
val words = lines.flatMap { line => line.split(" ") }
words.foreach { word => println(word) }
}
def groupByKey() {
val conf = new SparkConf()
.setAppName("groupByKey")
.setMaster("local")
val sc = new SparkContext(conf)
val scoreList = Array(Tuple2("class1", 80), Tuple2("class2", 75),
Tuple2("class1", 90), Tuple2("class2", 60))
val scores = sc.parallelize(scoreList, 1)
val groupedScores = scores.groupByKey()
groupedScores.foreach(score => {
println(score._1);
score._2.foreach { singleScore => println(singleScore) };
println("=============================")
})
}
def reduceByKey() {
val conf = new SparkConf()
.setAppName("groupByKey")
.setMaster("local")
val sc = new SparkContext(conf)
val scoreList = Array(Tuple2("class1", 80), Tuple2("class2", 75),
Tuple2("class1", 90), Tuple2("class2", 60))
val scores = sc.parallelize(scoreList, 1)
val totalScores = scores.reduceByKey(_ + _)
totalScores.foreach(classScore => println(classScore._1 + ": " + classScore._2))
}
def sortByKey() {
val conf = new SparkConf()
.setAppName("sortByKey")
.setMaster("local")
val sc = new SparkContext(conf)
val scoreList = Array(Tuple2(65, "leo"), Tuple2(50, "tom"),
Tuple2(100, "marry"), Tuple2(85, "jack"))
val scores = sc.parallelize(scoreList, 1)
val sortedScores = scores.sortByKey(false)
sortedScores.foreach(studentScore => println(studentScore._1 + ": " + studentScore._2))
}
def join() {
val conf = new SparkConf()
.setAppName("join")
.setMaster("local")
val sc = new SparkContext(conf)
val studentList = Array(
Tuple2(1, "leo"),
Tuple2(2, "jack"),
Tuple2(3, "tom"));
val scoreList = Array(
Tuple2(1, 100),
Tuple2(2, 90),
Tuple2(3, 60));
val students = sc.parallelize(studentList);
val scores = sc.parallelize(scoreList);
val studentScores = students.join(scores)
studentScores.foreach(studentScore => {
println("student id: " + studentScore._1);
println("student name: " + studentScore._2._1)
println("student socre: " + studentScore._2._2)
println("=======================================")
})
}
def cogroup() {
}
}
