Spark常用transformation和action操作

51 阅读 0 评论 34 点赞

我是靠谱客的博主瘦瘦睫毛膏，最近开发中收集的这篇文章主要介绍Spark常用transformation和action操作，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

1，场景

     spark的transformation和action操作乃spark的核心，让我们走进他们，领略他们的魅力！

2，transformation和action介绍

   Spark支持两种RDD操作：transformation和action。transformation操作会针对已有的RDD创建一个新的RDD；而action则主要是对RDD进行最后的操作，比如遍历、reduce、保存到文件等，并可以返回结果给Driver程序。

2.1，常用transformation:

操作	介绍
map	将RDD中的每个元素传入自定义函数，获取一个新的元素，然后用新的元素组成新的RDD
filter	对RDD中每个元素进行判断，如果返回true则保留，返回false则剔除
flatMap	与map类似，但是对每个元素都可以返回一个或多个新元素
gropuByKey	根据key进行分组，每个key对应一个Iterable
reduceByKey	对每个key对应的value进行reduce操作
sortByKey	对每个key对应的value进行排序操作
join	对两个包含对的RDD进行join操作，每个key join上的pair，都会传入自定义函数进行处理
cogroup	同join，但是是每个key对应的Iterable都会传入自定义函数进行处理

2.2,常用action介绍

操作	介绍
reduce	将RDD中的所有元素进行聚合操作。第一个和第二个元素聚合，值与第三个元素聚合，值与第四个元素聚合，以此类推
collect	将RDD中所有元素获取到本地客户端
count	获取RDD元素总数
take(n)	获取RDD中前n个元素
saveAsTextFile	将RDD元素保存到文件中，对每个元素调用toString方法
countByKey	对每个key对应的值进行count计数
foreach	遍历RDD中的每个元素

3,实战代码

package cn.spark.study.core;

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操作实战
 * @author Administrator
 *
 */
@SuppressWarnings(value = {"unused", "unchecked"})
public class TransformationOperation {

    public static void main(String[] args) {
        // map();
        // filter();
        // flatMap();
        // groupByKey();
        // reduceByKey();
        // sortByKey();
        // join();
        cogroup();
    }

    /**
     * map算子案例：将集合中每一个元素都乘以2
     */
    private static void map() {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("map")
                .setMaster("local");
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 构造集合
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);

        // 创建初始RDD
        JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

        // 使用map算子，将集合中的每个元素都乘以2
                JavaRDD<Integer> multipleNumberRDD = numberRDD.map(

                new Function<Integer, Integer>() {

                    private static final long serialVersionUID = 1L;


                    @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);  
            }

        });

        // 关闭JavaSparkContext
        sc.close();
    }

    /**
     * filter算子案例：过滤集合中的偶数
     */
    private static void filter() {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("filter")
                .setMaster("local");
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 模拟集合
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        // 创建初始RDD
        JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

        // 对初始RDD执行filter算子，过滤出其中的偶数

        JavaRDD<Integer> evenNumberRDD = numberRDD.filter(

                new Function<Integer, Boolean>() {

                    private static final long serialVersionUID = 1L;


                    @Override
                    public Boolean call(Integer v1) throws Exception {
                        return v1 % 2 == 0;
                    }

                });

        // 打印新的RDD
        evenNumberRDD.foreach(new VoidFunction<Integer>() {

            private static final long serialVersionUID = 1L;

            @Override
            public void call(Integer t) throws Exception {
                System.out.println(t);
            }

        });

        // 关闭JavaSparkContext
        sc.close();
    }

    /**
     * flatMap案例：将文本行拆分为多个单词
     */
    private static void flatMap() {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("flatMap")  
                .setMaster("local");  
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);

        // 构造集合
        List<String> lineList = Arrays.asList("hello you", "hello me", "hello world");  

        // 创建RDD
        JavaRDD<String> lines = sc.parallelize(lineList);

        // 对RDD执行flatMap算子，将每一行文本，拆分为多个单词
                JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

            private static final long serialVersionUID = 1L;

                        @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);
            }
        });

        // 关闭JavaSparkContext
        sc.close();
    }

    /**
     * groupByKey案例：按照班级对成绩进行分组
     */
    private static void groupByKey() {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("groupByKey")  
                .setMaster("local");
        // 创建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));

        // 创建JavaPairRDD
        JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);

        // 针对scores RDD，执行groupByKey算子，对每个班级的成绩进行分组
                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("==============================");   
            }

        });

        // 关闭JavaSparkContext
        sc.close();
    }

    /**
     * reduceByKey案例：统计每个班级的总分
     */
    private static void reduceByKey() {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("reduceByKey")  
                .setMaster("local");
        // 创建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));

        // 创建JavaPairRDD
        JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);

        // 针对scores RDD，执行reduceByKey算子
                JavaPairRDD<String, Integer> totalScores = scores.reduceByKey(

                new Function2<Integer, Integer, Integer>() {

                    private static final long serialVersionUID = 1L;

                                        @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);   
            }

        });

        // 关闭JavaSparkContext
        sc.close();
    }

    /**
     * sortByKey案例：按照学生分数进行排序
     */
    private static void sortByKey() {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("sortByKey")  
                .setMaster("local");
        // 创建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"));

        // 创建RDD
        JavaPairRDD<Integer, String> scores = sc.parallelizePairs(scoreList);

        // 对scores RDD执行sortByKey算子
                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);  
            }

        });

        // 关闭JavaSparkContext
        sc.close();
    }

    /**
     * join案例：打印学生成绩
     */
    private static void join() {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("join")  
                .setMaster("local");
        // 创建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));

        // 并行化两个RDD
        JavaPairRDD<Integer, String> students = sc.parallelizePairs(studentList);
        JavaPairRDD<Integer, Integer> scores = sc.parallelizePairs(scoreList);

        // 使用join算子关联两个RDD
                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("===============================");   
                    }

                });

        // 关闭JavaSparkContext
        sc.close();
    }

    /**
     * cogroup案例：打印学生成绩
     */
    private static void cogroup() {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("cogroup")  
                .setMaster("local");
        // 创建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));

        // 并行化两个RDD
        JavaPairRDD<Integer, String> students = sc.parallelizePairs(studentList);
        JavaPairRDD<Integer, Integer> scores = sc.parallelizePairs(scoreList);

        // cogroup与join不同
        // 相当于是，一个key join上的所有value，都给放到一个Iterable里面去了 

        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);
                        System.out.println("===============================");   
                    }

                });

        // 关闭JavaSparkContext
        sc.close();
    }## 标题 ##

}