spark基本操作 java 版-低调大师

spark基本操作 java 版

2017-11-11 462

1.map算子

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>() {  
            @Override  
            public Integer call(Integer v1) throws Exception {  
                return v1 * 2;  
            }  
        });  
        //打印新的RDD  
        multipleNumberRDD.foreach(new VoidFunction<Integer>() {  
            @Override  
            public void call(Integer t) throws Exception {  
                System.out.println(t);  
            }  
        });  
        //关闭JavasparkContext  
        sc.close();  
    }

2.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);  

        //对集合使用filter算子，过滤出集合中的偶数  
        JavaRDD<Integer> evenNumberRDD = numberRDD.filter(new Function<Integer, Boolean>() {  
            @Override  
            public Boolean call(Integer v1) throws Exception {  
                return v1%2==0;  
            }  
        });  
        evenNumberRDD.foreach(new VoidFunction<Integer>() {  
            @Override  
            public void call(Integer t) throws Exception {  
                System.out.println(t);  
            }  

        });  
        sc.close();  
    }

3.flatMap算子

Spark 中 map函数会对每一条输入进行指定的操作，然后为每一条输入返回一个对象；

而flatMap函数则是两个操作的集合——正是“先映射后扁平化”：

操作1：同map函数一样：对每一条输入进行指定的操作，然后为每一条输入返回一个对象

操作2：最后将所有对象合并为一个对象

private static void flatMap() {  
        SparkConf conf = new SparkConf()  
            .setAppName("flatMap")  
            .setMaster("local");  

        JavaSparkContext sc = new JavaSparkContext(conf);  

        List<String> lineList = Arrays.asList("hello you","hello me","hello world");  

        JavaRDD<String> lines = sc.parallelize(lineList);  

        //对RDD执行flatMap算子，将每一行文本，拆分为多个单词  
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {  
            //在这里，传入第一行，hello,you  
            //返回的是一个Iterable<String>(hello,you)  
            @Override  
            public Iterable<String> call(String t) throws Exception {  
                return Arrays.asList(t.split(" "));  
            }  
        });  

        words.foreach(new VoidFunction<String>() {  
            @Override  
            public void call(String t) throws Exception {  
                System.out.println(t);  
            }  
        });  
        sc.close();  
    }

4.groupByKey算子

private static void groupByKey() {  
        SparkConf conf = new SparkConf()  
                .setAppName("groupByKey")  
                .setMaster("local");  
        JavaSparkContext sc = new JavaSparkContext(conf);  
        List<Tuple2<String, Integer>> scoreList = Arrays.asList(  
                new Tuple2<String, Integer>("class1", 80),  
                new Tuple2<String, Integer>("class2", 90),  
                new Tuple2<String, Integer>("class1", 97),  
                new Tuple2<String, Integer>("class2", 89));  

        JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);  
        //针对scoresRDD，执行groupByKey算子，对每个班级的成绩进行分组  
        //相当于是，一个key join上的所有value，都放到一个Iterable里面去了  
        JavaPairRDD<String, Iterable<Integer>> groupedScores = scores.groupByKey();  
        groupedScores.foreach(new VoidFunction<Tuple2<String,Iterable<Integer>>>() {  

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

5.reduceByKey算子

private static void reduceByKey() {  
        SparkConf conf = new SparkConf()  
                .setAppName("reduceByKey")  
                .setMaster("local");  

        JavaSparkContext sc = new JavaSparkContext(conf);  
        List<Tuple2<String, Integer>> scoreList = Arrays.asList(  
                new Tuple2<String, Integer>("class1", 80),  
                new Tuple2<String, Integer>("class2", 90),  
                new Tuple2<String, Integer>("class1", 97),  
                new Tuple2<String, Integer>("class2", 89));  

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

        //reduceByKey算法返回的RDD，还是JavaPairRDD<key,value>  
        JavaPairRDD<String, Integer> totalScores = scores.reduceByKey(new Function2<Integer, Integer, Integer>() {  
            @Override  
            public Integer call(Integer v1, Integer v2) throws Exception {  
                return v1 + v2;  
            }  
        });  

        totalScores.foreach(new VoidFunction<Tuple2<String,Integer>>() {  
            @Override  
            public void call(Tuple2<String, Integer> t) throws Exception {  
                System.out.println(t._1 + ":" + t._2);  

            }  
        });  
        sc.close();  
    }

6.sortByKey算子

private static void sortByKey() {  
        SparkConf conf = new SparkConf()  
                .setAppName("sortByKey")  
                .setMaster("local");  

        JavaSparkContext sc = new JavaSparkContext(conf);  

        List<Tuple2<Integer, String>> scoreList = Arrays.asList(  
                new Tuple2<Integer, String>(78, "marry"),  
                new Tuple2<Integer, String>(89, "tom"),  
                new Tuple2<Integer, String>(72, "jack"),  
                new Tuple2<Integer, String>(86, "leo"));  

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

        JavaPairRDD<Integer, String> sortedScores = scores.sortByKey();  
        sortedScores.foreach(new VoidFunction<Tuple2<Integer,String>>() {  
            @Override  
            public void call(Tuple2<Integer, String> t) throws Exception {  
                System.out.println(t._1 + ":" + t._2);  
            }  
        });  
        sc.close();  
    }

7.join算子
join算子用于关联两个RDD，join以后，会根据key进行join，并返回JavaPairRDD。JavaPairRDD的第一个泛型类型是之前两个JavaPairRDD的key类型，因为通过key进行join的。第二个泛型类型，是Tuple2<v1, v2>的类型，Tuple2的两个泛型分别为原始RDD的value的类型

private static void join() {  
        SparkConf conf = new SparkConf()  
                .setAppName("join")  
                .setMaster("local");  

        JavaSparkContext sc = new JavaSparkContext(conf);  

        List<Tuple2<Integer, String>> studentList = Arrays.asList(  
                new Tuple2<Integer, String>(1, "tom"),  
                new Tuple2<Integer, String>(2, "jack"),  
                new Tuple2<Integer, String>(3, "marry"),  
                new Tuple2<Integer, String>(4, "leo"));  

        List<Tuple2<Integer, Integer>> scoreList = Arrays.asList(  
                new Tuple2<Integer, Integer>(1, 78),  
                new Tuple2<Integer, Integer>(2, 87),  
                new Tuple2<Integer, Integer>(3, 89),  
                new Tuple2<Integer, Integer>(4, 98));  

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

        //使用join算子关联两个RDD  
        //join以后，会根据key进行join，并返回JavaPairRDD  
        //JavaPairRDD的第一个泛型类型，之前两个JavaPairRDD的key类型，因为通过key进行join的  
        //第二个泛型类型，是Tuple2<v1, v2>的类型，Tuple2的两个泛型分别为原始RDD的value的类型  
        JavaPairRDD<Integer, Tuple2<String, Integer>> studentScores = students.join(scores);  

        //打印  
        studentScores.foreach(new VoidFunction<Tuple2<Integer,Tuple2<String,Integer>>>() {  
            @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("==========================");  
            }  
        });  
        sc.close();  
    }

更深的方法参见：
http://blog.csdn.net/liulingyuan6/article/details/53397780
http://blog.csdn.net/liulingyuan6/article/details/53410832
https://www.2cto.com/net/201608/543044.html

转自： http://blog.csdn.net/sunhaoning/article/details/70505718

本文转自whk66668888 51CTO博客，原文链接:http://blog.51cto.com/12597095/2063767

微信关注我们

原文链接：https://yq.aliyun.com/articles/553769

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

Spark中RDD操作

Transformations(转换) Transformation 说明 map(func) 根据传入的函数处理原有的RDD对象中每一个元素，每一个新元素处理完成后返回一个对象,这些新对象组装得到一个新的RDD,新的RDD和旧的RDD元素都是一一对应的 filter(func) 根据传入的函数来过滤RDD中每一个元素,通过过滤条件的的元素组成一个新的RDD flatMap(func) 先进行map操作,然后把map操作得到的结果合并为一个对象,假如map操作返回的是Array[Array[String]],那flatMap操作得到的应该是Array[String],自动将多个字符串数组合并为一个,另外一个意义是一个旧的RDD元素可以生成多个新元素，一对多的关系 mapPartitions(func) 可以将其看成map,但是他处理的是每个单独分区中的数据,然后把各个分区的值合并 mapPartitionsWithIndex(func) 把分区的索引值(index)交给输入函数处理 sample(withReplacement, fraction, seed) 取样函数,分放回和不放回...

2017-11-12

562

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 curl-XPUT "http://localhost:9200/sho...

2017-11-12

827

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。