Spark算子实战Java版,学到了

Posted Java鱼仔

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark算子实战Java版,学到了相关的知识,希望对你有一定的参考价值。

(一)概述

算子从功能上可以分为Transformations转换算子和Action行动算子。转换算子用来做数据的转换操作,比如map、flatMap、reduceByKey等都是转换算子,这类算子通过懒加载执行。行动算子的作用是触发执行,比如foreach、collect、count等都是行动算子,只有程序运行到行动算子时,转换算子才会去执行。

本文将介绍开发过程中常用的转换算子和行动算子,Spark代码基于Java编写,前置代码如下:

public class SparkTransformationTest 
    public static void main(String[] args) 
        // 前置准备
        SparkConf conf = new SparkConf();
        conf.setMaster("local[1]");
        conf.setAppName("SPARK ES");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<String> javaRdd = sc.parallelize(Arrays.asList("a", "b", "c", "d", "e"));
    

(二)转换算子

map

map(func):通过函数func传递的每个元素,返回一个新的RDD。

JavaRDD<Object> map = javaRdd.map((Function<String, Object>) 
        item -> "new" + item);
map.foreach(x -> System.out.println(x));

返回一个新的RDD,数据是newa、newb、newc、newd、newe

filter

filter(func):筛选通过func处理后返回 true 的元素,返回一个新的RDD。

JavaRDD<String> filter = javaRdd.filter(item -> item.equals("a") || item.equals("b"));
filter.foreach(x -> System.out.println(x));

返回的新RDD数据是a和b。

flatMap

flatMap(func):类似于 map,但每个输入项可以映射到 0 个或更多输出项。

JavaRDD<String> javaRdd = sc.parallelize(Arrays.asList("a,b", "c,d,e", "f,g"));
JavaRDD<String> flatMap = javaRdd.flatMap((FlatMapFunction<String, String>) 
        s -> Arrays.asList(s.split(",")).iterator());
flatMap.foreach(x -> System.out.println(x));

入参只有3个,经过flatMap映射后返回了长度为7的RDD。

mapPartitions

mapPartitions(func):类似于map,但该函数是在RDD每个partition上单独运行,因此入参会是Iterator<Object>

JavaRDD<String> mapPartitions = javaRdd.mapPartitions((FlatMapFunction<Iterator<String>, String>) stringIterator -> 
    ArrayList<String> list = new ArrayList<>();
    while (stringIterator.hasNext()) 
        list.add(stringIterator.next());
    
    return list.iterator();
);
mapPartitions.foreach(x -> System.out.println(x));

除了是对Iterator进行处理之外,其他的都和map一样。

union

union(otherDataset):返回一个新数据集,包含两个数据集合的并集。

JavaRDD<String> newJavaRdd = sc.parallelize(Arrays.asList("1", "2", "3", "4", "5"));
JavaRDD<String> unionRdd = javaRdd.union(newJavaRdd);
unionRdd.foreach(x-> System.out.println(x));

intersection

intersection(otherDataset):返回一个新数据集,包含两个数据集合的交集。

JavaRDD<String> newJavaRdd = sc.parallelize(Arrays.asList("a", "b", "3", "4", "5"));
JavaRDD<String> intersectionRdd = javaRdd.intersection(newJavaRdd);
intersectionRdd.foreach(x-> System.out.println(x));

groupByKey

groupByKey ([ numPartitions ]):在 (K, V) 对的数据集上调用时,返回 (K, Iterable) 对的数据集,可以传递一个可选numPartitions参数来设置不同数量的任务。

这里需要了解Java中的另外一种RDD,JavaPairRDD。JavaPairRDD是一种key-value类型的RDD,groupByKey就是针对JavaPairRDD的API。

JavaRDD<String> rdd = sc.parallelize(Arrays.asList("a:1", "a:2", "b:1", "c:3"));
JavaPairRDD<String, Integer> javaPairRDD = rdd.mapToPair(s -> 
    String[] split = s.split(":");
    return new Tuple2<>(split[0], Integer.parseInt(split[1]));
);
JavaPairRDD<String, Iterable<Integer>> groupByKey = javaPairRDD.groupByKey();
groupByKey.foreach(x -> System.out.println(x._1()+x._2()));

最终输出结果:

a[1, 2]
b[1]
c[3]

reduceByKey

reduceByKey(func, [numPartitions]):在 (K, V) 对的数据集上调用时,返回 (K, V) 对的数据集,其中每个键的值使用给定的 reduce 函数func聚合。和groupByKey不同的地方在于reduceByKey对value进行了聚合处理。

JavaRDD<String> rdd = sc.parallelize(Arrays.asList("a:1", "a:2", "b:1", "c:3"));
JavaPairRDD<String, Integer> javaPairRDD = rdd.mapToPair(s -> 
    String[] split = s.split(":");
    return new Tuple2<>(split[0], Integer.parseInt(split[1]));
);
JavaPairRDD<String, Integer> reduceRdd = javaPairRDD.reduceByKey((Function2<Integer, Integer, Integer>) (x, y) -> x + y);
reduceRdd.foreach(x -> System.out.println(x._1()+x._2()));

最终输出结果:

a3
b1
c3

aggregateByKey

aggregateByKey(zeroValue)(seqOp, combOp, [numPartitions]):aggregateByKey这个算子相比上面这些会复杂很多,主要参数有zeroValue、seqOp、combOp,numPartitions可选。

zeroValue是该算子设置的初始值,seqOp函数是将rdd中的value值和zeroValue进行处理,combOp是将相同key的数据进行处理。

JavaRDD<String> rdd = sc.parallelize(Arrays.asList("a:1", "a:2", "b:1", "c:3"));
JavaPairRDD<String, Integer> javaPairRDD = rdd.mapToPair(s -> 
    String[] split = s.split(":");
    return new Tuple2<>(split[0], Integer.parseInt(split[1]));
);
JavaPairRDD<String, Integer> aggregateRdd = javaPairRDD.aggregateByKey(1,
        // seqOp函数中的第一个入参是 zeroValue,第二个入参是rdd的value,这里对所有的value+1
        (Function2<Integer, Integer, Integer>) (v1, v2) -> v1 + v2,
        // combOp函数是对同一个key的value进行处理,这里对相同key的value进行相加
        (Function2<Integer, Integer, Integer>) (v1, v2) -> v1 + v2);
aggregateRdd.foreach(x -> System.out.println(x._1()+":"+x._2()));

最终输出结果如下:

a:4
b:2
c:4

(三)行动算子

reduce

reduce(func):使用函数func聚合数据集的元素(它接受两个参数并返回一个)。
下面这段代码对所有rdd进行相加:

String reduce = javaRdd.reduce((Function2<String, String, String>) (v1, v2) -> 
    System.out.println(v1 + ":" + v2);
    return v1+v2;
);
System.out.println("result:"+reduce);

最终结果如下,从结果可以看出,每次对v1都是上一次reduce运行之后的结果:

a:b
ab:c
abc:d
abcd:e
result:abcde

collect()

collect():将driver中的所有元素数据通过集合的方式返回,适合小数据量的场景,大数据量会导致内存溢出。

JavaRDD<String> javaRdd = sc.parallelize(Arrays.asList("a", "b", "c", "d", "e"));
List<String> collect = javaRdd.collect();

count()

count():返回一个RDD中元素的数量。

JavaRDD<String> javaRdd = sc.parallelize(Arrays.asList("a", "b", "c", "d", "e"));
long count = javaRdd.count();

first()

first():返回第一个元素。

JavaRDD<String> javaRdd = sc.parallelize(Arrays.asList("a", "b", "c", "d", "e"));
String first = javaRdd.first();

take

take(n):返回前N个元素,take(1)=first()。

JavaRDD<String> javaRdd = sc.parallelize(Arrays.asList("a", "b", "c", "d", "e"));
List<String> take = javaRdd.take(3);

takeOrdered

takeOrdered(n, [ordering]):返回自然排序的前N个元素,或者指定排序方法后的前N个元素。首先写一个排序类。

public class MyComparator implements Comparator<String>, Serializable 
    @Override
    public int compare(String o1, String o2) 
        return o2.compareTo(o1);
    

接着是调用方式:

JavaRDD<String> javaRdd = sc.parallelize(Arrays.asList("a", "b", "c", "d", "e"));
List<String> take = javaRdd.takeOrdered(3, new MyComparator());

foreach

foreach(func):该函数对数据集的每个RDD运行func函数,foreach算子在上面的代码中已经使用到,这里不再做代码案例展示。

(四)总结

Spark的开发可以用Java或者Scala,Spark本身使用Scala编写,具体使用哪种语言进行开发需要根据项目情况考虑时间和学习成本。具体的API都可以在Spark官网查询:https://spark.apache.org/docs/2.3.0/rdd-programming-guide.html

以上是关于Spark算子实战Java版,学到了的主要内容,如果未能解决你的问题,请参考以下文章

Spark算子实战Java版,学到了

Spark中的各种action算子操作(java版)

groupByKey,reduceByKey,sortByKey算子-Java&Python版Spark

Spark RDD算子实战

Spark Transformation 算子

Spark算子---实战应用