V2AS问路
27.Spark中transformation的介绍
阅读原文时间：2023年07月09日阅读：2

Spark支持两种RDD操作：transformation和action。transformation操作会针对已有的RDD创建一个新的RDD；

而action则主要是对RDD进行最后的操作，比如遍历、reduce、保存到文件等，并可以返回结果给Driver程序。

例如，map就是一种transformation操作，它用于将已有RDD的每个元素传入一个自定义的函数，并获取一个新的元素，然后将所有的新元素组成一个新的RDD。

而reduce就是一种action操作，它用于对RDD中的所有元素进行聚合操作，并获取一个最终的结果，然后返回给Driver程序。

transformation的特点就是lazy特性。lazy特性指的是，如果一个spark应用中只定义了transformation操作，那么即使你执行该应用，

这些操作也不会执行。也就是说，transformation是不会触发spark程序的执行的，它们只是记录了对RDD所做的操作，但是不会自发的执行。

只有当transformation之后，接着执行了一个action操作，那么所有的transformation才会执行。Spark通过这种lazy特性，来进行底层的spark应用执行的优化，避免产生过多中间结果。

action操作执行，会触发一个spark job的运行，从而触发这个action之前所有的transformation的执行。这是action的特性。

常用transformation介绍

1、map：将集合中每个元素乘以2

2、filter：过滤出集合中的偶数

3、flatMap：将行拆分为单词

4、groupByKey：将每个班级的成绩进行分组

5、reduceByKey：统计每个班级的总分

6、sortByKey：将学生分数进行排序

7、join：打印每个学生的成绩

8、cogroup：打印每个学生的成绩

map：将集合中每个元素乘以2

新建TransforDemo类

package com.it19gong.sparkproject;

import java.util.Arrays;
import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;

public class TransforDemo {

public static void main(String\[\] args) {  
    map();

}

private static void map() {  
    //创建SparkConf  
    SparkConf conf = new SparkConf().setAppName("map").setMaster("local");  
    // 创建JavaSparkContext  
    JavaSparkContext sc = new JavaSparkContext(conf);  
    //构建集合  
    List numbers = Arrays.asList(,,,,);  
    //并行化集合，创建初始RDD  
    JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

       // 使用map算子，将集合中的每个元素都乘以2  
    // map算子，是对任何类型的RDD，都可以调用的  
    // 在java中，map算子接收的参数是Function对象  
    // 创建的Function对象，一定会让你设置第二个泛型参数，这个泛型类型，就是返回的新元素的类型  
    // 同时call()方法的返回类型，也必须与第二个泛型类型同步  
    // 在call()方法内部，就可以对原始RDD中的每一个元素进行各种处理和计算，并返回一个新的元素  
    // 所有新的元素就会组成一个新的RDD

    // 所有新的元素就会组成一个新的RDD  
    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 \* ;

        }  
    });

     multipleNumberRDD.foreach(new VoidFunction<Integer>() {

        private static final long serialVersionUID = 1L;

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

}

}

运行代码

添加filter()方法

package com.it19gong.sparkproject;

import java.util.Arrays;
import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;

public class TransforDemo {

public static void main(String\[\] args) {  
    //map();  
    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(, , , , , , , , , );

            // 并行化集合，创建初始RDD  
            JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

            // 对初始RDD执行filter算子，过滤出其中的偶数  
            // filter算子，传入的也是Function，其他的使用注意点，实际上和map是一样的  
            // 但是，唯一的不同，就是call()方法的返回类型是Boolean  
            // 每一个初始RDD中的元素，都会传入call()方法，此时你可以执行各种自定义的计算逻辑  
            // 来判断这个元素是否是你想要的  
            // 如果你想在新的RDD中保留这个元素，那么就返回true；否则，不想保留这个元素，返回false

            JavaRDD<Integer> evenNumberRDD = numberRDD.filter(

                    new Function<Integer, Boolean>() {

                        private static final long serialVersionUID = 1L;

                        // 在这里，1到10，都会传入进来  
                        // 但是根据我们的逻辑，只有2,4,6,8,10这几个偶数，会返回true  
                        // 所以，只有偶数会保留下来，放在新的RDD中  
                        @Override  
                        public Boolean call(Integer v1) throws Exception {  
                            return v1 %  == ;  
                        }

                    });

            // 打印新的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();  
        }

private static void map() {  
    //创建SparkConf  
    SparkConf conf = new SparkConf().setAppName("map").setMaster("local");  
    // 创建JavaSparkContext  
    JavaSparkContext sc = new JavaSparkContext(conf);  
    //构建集合  
    List numbers = Arrays.asList(,,,,);  
    //并行化集合，创建初始RDD  
    JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

       // 使用map算子，将集合中的每个元素都乘以2  
    // map算子，是对任何类型的RDD，都可以调用的  
    // 在java中，map算子接收的参数是Function对象  
    // 创建的Function对象，一定会让你设置第二个泛型参数，这个泛型类型，就是返回的新元素的类型  
    // 同时call()方法的返回类型，也必须与第二个泛型类型同步  
    // 在call()方法内部，就可以对原始RDD中的每一个元素进行各种处理和计算，并返回一个新的元素  
    // 所有新的元素就会组成一个新的RDD

    // 所有新的元素就会组成一个新的RDD  
    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 \* ;

        }  
    });

     multipleNumberRDD.foreach(new VoidFunction<Integer>() {

        private static final long serialVersionUID = 1L;

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

}

}

运行filter()

flatMap案例

flatMap案例：将文本行拆分为多个单词

package com.it19gong.sparkproject;

import java.util.Arrays;
import java.util.List;
import org.apache.spark.SparkConf;
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.VoidFunction;

public class TransforDemo {

public static void main(String\[\] args) {  
    //map();  
    //filter();  
    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算子，将每一行文本，拆分为多个单词  
            // flatMap算子，在java中，接收的参数是FlatMapFunction  
            // 我们需要自己定义FlatMapFunction的第二个泛型类型，即，代表了返回的新元素的类型  
            // call()方法，返回的类型，不是U，而是Iterable<U>，这里的U也与第二个泛型类型相同  
            // flatMap其实就是，接收原始RDD中的每个元素，并进行各种逻辑的计算和处理，可以返回多个元素  
            // 多个元素，即封装在Iterable集合中，可以使用ArrayList等集合  
            // 新的RDD中，即封装了所有的新元素；也就是说，新的RDD的大小一定是 >= 原始RDD的大小  
            JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

                private static final long serialVersionUID = 1L;

                // 在这里会，比如，传入第一行，hello you  
                // 返回的是一个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);  
                }  
            });

            // 关闭JavaSparkContext  
            sc.close();

}

private static void filter() {  
            // 创建SparkConf  
            SparkConf conf = new SparkConf().setAppName("filter").setMaster("local");  
            // 创建JavaSparkContext  
            JavaSparkContext sc = new JavaSparkContext(conf);

            // 模拟集合  
            List<Integer> numbers = Arrays.asList(, , , , , , , , , );

            // 并行化集合，创建初始RDD  
            JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

            // 对初始RDD执行filter算子，过滤出其中的偶数  
            // filter算子，传入的也是Function，其他的使用注意点，实际上和map是一样的  
            // 但是，唯一的不同，就是call()方法的返回类型是Boolean  
            // 每一个初始RDD中的元素，都会传入call()方法，此时你可以执行各种自定义的计算逻辑  
            // 来判断这个元素是否是你想要的  
            // 如果你想在新的RDD中保留这个元素，那么就返回true；否则，不想保留这个元素，返回false

            JavaRDD<Integer> evenNumberRDD = numberRDD.filter(

                    new Function<Integer, Boolean>() {

                        private static final long serialVersionUID = 1L;

                        // 在这里，1到10，都会传入进来  
                        // 但是根据我们的逻辑，只有2,4,6,8,10这几个偶数，会返回true  
                        // 所以，只有偶数会保留下来，放在新的RDD中  
                        @Override  
                        public Boolean call(Integer v1) throws Exception {  
                            return v1 %  == ;  
                        }

                    });

            // 打印新的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();  
        }

private static void map() {  
    //创建SparkConf  
    SparkConf conf = new SparkConf().setAppName("map").setMaster("local");  
    // 创建JavaSparkContext  
    JavaSparkContext sc = new JavaSparkContext(conf);  
    //构建集合  
    List numbers = Arrays.asList(,,,,);  
    //并行化集合，创建初始RDD  
    JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

       // 使用map算子，将集合中的每个元素都乘以2  
    // map算子，是对任何类型的RDD，都可以调用的  
    // 在java中，map算子接收的参数是Function对象  
    // 创建的Function对象，一定会让你设置第二个泛型参数，这个泛型类型，就是返回的新元素的类型  
    // 同时call()方法的返回类型，也必须与第二个泛型类型同步  
    // 在call()方法内部，就可以对原始RDD中的每一个元素进行各种处理和计算，并返回一个新的元素  
    // 所有新的元素就会组成一个新的RDD

    // 所有新的元素就会组成一个新的RDD  
    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 \* ;

        }  
    });

     multipleNumberRDD.foreach(new VoidFunction<Integer>() {

        private static final long serialVersionUID = 1L;

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

}

}

运行代码

GroupBYkey案例

将每个班级的成绩进行分组

package com.it19gong.sparkproject;

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.VoidFunction;

import scala.Tuple2;

public class TransforDemo {

public static void main(String\[\] args) {  
    //map();  
    //filter();  
    //flatMap() ;  
     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", ),  
                    new Tuple2<String, Integer>("class2", ),  
                    new Tuple2<String, Integer>("class1", ),  
                    new Tuple2<String, Integer>("class2", ));

            // 并行化集合，创建JavaPairRDD  
            JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);

            // 针对scores RDD，执行groupByKey算子，对每个班级的成绩进行分组  
            // groupByKey算子，返回的还是JavaPairRDD  
            // 但是，JavaPairRDD的第一个泛型类型不变，第二个泛型类型变成Iterable这种集合类型  
            // 也就是说，按照了key进行分组，那么每个key可能都会有多个value，此时多个value聚合成了Iterable  
            // 那么接下来，我们是不是就可以通过groupedScores这种JavaPairRDD，很方便地处理某个分组内的数据  
            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();

}

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算子，将每一行文本，拆分为多个单词  
            // flatMap算子，在java中，接收的参数是FlatMapFunction  
            // 我们需要自己定义FlatMapFunction的第二个泛型类型，即，代表了返回的新元素的类型  
            // call()方法，返回的类型，不是U，而是Iterable<U>，这里的U也与第二个泛型类型相同  
            // flatMap其实就是，接收原始RDD中的每个元素，并进行各种逻辑的计算和处理，可以返回多个元素  
            // 多个元素，即封装在Iterable集合中，可以使用ArrayList等集合  
            // 新的RDD中，即封装了所有的新元素；也就是说，新的RDD的大小一定是 >= 原始RDD的大小  
            JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

                private static final long serialVersionUID = 1L;

                // 在这里会，比如，传入第一行，hello you  
                // 返回的是一个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);  
                }  
            });

            // 关闭JavaSparkContext  
            sc.close();

}

private static void filter() {  
            // 创建SparkConf  
            SparkConf conf = new SparkConf().setAppName("filter").setMaster("local");  
            // 创建JavaSparkContext  
            JavaSparkContext sc = new JavaSparkContext(conf);

            // 模拟集合  
            List<Integer> numbers = Arrays.asList(, , , , , , , , , );

            // 并行化集合，创建初始RDD  
            JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

            // 对初始RDD执行filter算子，过滤出其中的偶数  
            // filter算子，传入的也是Function，其他的使用注意点，实际上和map是一样的  
            // 但是，唯一的不同，就是call()方法的返回类型是Boolean  
            // 每一个初始RDD中的元素，都会传入call()方法，此时你可以执行各种自定义的计算逻辑  
            // 来判断这个元素是否是你想要的  
            // 如果你想在新的RDD中保留这个元素，那么就返回true；否则，不想保留这个元素，返回false

            JavaRDD<Integer> evenNumberRDD = numberRDD.filter(

                    new Function<Integer, Boolean>() {

                        private static final long serialVersionUID = 1L;

                        // 在这里，1到10，都会传入进来  
                        // 但是根据我们的逻辑，只有2,4,6,8,10这几个偶数，会返回true  
                        // 所以，只有偶数会保留下来，放在新的RDD中  
                        @Override  
                        public Boolean call(Integer v1) throws Exception {  
                            return v1 %  == ;  
                        }

                    });

            // 打印新的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();  
        }

private static void map() {  
    //创建SparkConf  
    SparkConf conf = new SparkConf().setAppName("map").setMaster("local");  
    // 创建JavaSparkContext  
    JavaSparkContext sc = new JavaSparkContext(conf);  
    //构建集合  
    List numbers = Arrays.asList(,,,,);  
    //并行化集合，创建初始RDD  
    JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

       // 使用map算子，将集合中的每个元素都乘以2  
    // map算子，是对任何类型的RDD，都可以调用的  
    // 在java中，map算子接收的参数是Function对象  
    // 创建的Function对象，一定会让你设置第二个泛型参数，这个泛型类型，就是返回的新元素的类型  
    // 同时call()方法的返回类型，也必须与第二个泛型类型同步  
    // 在call()方法内部，就可以对原始RDD中的每一个元素进行各种处理和计算，并返回一个新的元素  
    // 所有新的元素就会组成一个新的RDD

    // 所有新的元素就会组成一个新的RDD  
    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 \* ;

        }  
    });

     multipleNumberRDD.foreach(new VoidFunction<Integer>() {

        private static final long serialVersionUID = 1L;

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

}

}

运行代码

SortByKey 案例

将学生分数进行排序

package com.it19gong.sparkproject;

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.VoidFunction;

import scala.Tuple2;

public class TransforDemo {

public static void main(String\[\] args) {  
    //map();  
    //filter();  
    //flatMap() ;  
    // groupByKey();  
    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>(, "leo"),  
            new Tuple2<Integer, String>(, "tom"),  
            new Tuple2<Integer, String>(, "marry"),  
            new Tuple2<Integer, String>(, "jack"));

    // 并行化集合，创建RDD  
    JavaPairRDD<Integer, String> scores = sc.parallelizePairs(scoreList);

    // 对scores RDD执行sortByKey算子  
    // sortByKey其实就是根据key进行排序，可以手动指定升序，或者降序  
    // 返回的，还是JavaPairRDD，其中的元素内容，都是和原始的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);  
        }

    });

    // 关闭JavaSparkContext  
    sc.close();

}

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", ),  
                    new Tuple2<String, Integer>("class2", ),  
                    new Tuple2<String, Integer>("class1", ),  
                    new Tuple2<String, Integer>("class2", ));

            // 并行化集合，创建JavaPairRDD  
            JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);

            // 针对scores RDD，执行groupByKey算子，对每个班级的成绩进行分组  
            // groupByKey算子，返回的还是JavaPairRDD  
            // 但是，JavaPairRDD的第一个泛型类型不变，第二个泛型类型变成Iterable这种集合类型  
            // 也就是说，按照了key进行分组，那么每个key可能都会有多个value，此时多个value聚合成了Iterable  
            // 那么接下来，我们是不是就可以通过groupedScores这种JavaPairRDD，很方便地处理某个分组内的数据  
            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();

}

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算子，将每一行文本，拆分为多个单词  
            // flatMap算子，在java中，接收的参数是FlatMapFunction  
            // 我们需要自己定义FlatMapFunction的第二个泛型类型，即，代表了返回的新元素的类型  
            // call()方法，返回的类型，不是U，而是Iterable<U>，这里的U也与第二个泛型类型相同  
            // flatMap其实就是，接收原始RDD中的每个元素，并进行各种逻辑的计算和处理，可以返回多个元素  
            // 多个元素，即封装在Iterable集合中，可以使用ArrayList等集合  
            // 新的RDD中，即封装了所有的新元素；也就是说，新的RDD的大小一定是 >= 原始RDD的大小  
            JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

                private static final long serialVersionUID = 1L;

                // 在这里会，比如，传入第一行，hello you  
                // 返回的是一个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);  
                }  
            });

            // 关闭JavaSparkContext  
            sc.close();

}

private static void filter() {  
            // 创建SparkConf  
            SparkConf conf = new SparkConf().setAppName("filter").setMaster("local");  
            // 创建JavaSparkContext  
            JavaSparkContext sc = new JavaSparkContext(conf);

            // 模拟集合  
            List<Integer> numbers = Arrays.asList(, , , , , , , , , );

            // 并行化集合，创建初始RDD  
            JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

            // 对初始RDD执行filter算子，过滤出其中的偶数  
            // filter算子，传入的也是Function，其他的使用注意点，实际上和map是一样的  
            // 但是，唯一的不同，就是call()方法的返回类型是Boolean  
            // 每一个初始RDD中的元素，都会传入call()方法，此时你可以执行各种自定义的计算逻辑  
            // 来判断这个元素是否是你想要的  
            // 如果你想在新的RDD中保留这个元素，那么就返回true；否则，不想保留这个元素，返回false

            JavaRDD<Integer> evenNumberRDD = numberRDD.filter(

                    new Function<Integer, Boolean>() {

                        private static final long serialVersionUID = 1L;

                        // 在这里，1到10，都会传入进来  
                        // 但是根据我们的逻辑，只有2,4,6,8,10这几个偶数，会返回true  
                        // 所以，只有偶数会保留下来，放在新的RDD中  
                        @Override  
                        public Boolean call(Integer v1) throws Exception {  
                            return v1 %  == ;  
                        }

                    });

            // 打印新的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();  
        }

private static void map() {  
    //创建SparkConf  
    SparkConf conf = new SparkConf().setAppName("map").setMaster("local");  
    // 创建JavaSparkContext  
    JavaSparkContext sc = new JavaSparkContext(conf);  
    //构建集合  
    List numbers = Arrays.asList(,,,,);  
    //并行化集合，创建初始RDD  
    JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

       // 使用map算子，将集合中的每个元素都乘以2  
    // map算子，是对任何类型的RDD，都可以调用的  
    // 在java中，map算子接收的参数是Function对象  
    // 创建的Function对象，一定会让你设置第二个泛型参数，这个泛型类型，就是返回的新元素的类型  
    // 同时call()方法的返回类型，也必须与第二个泛型类型同步  
    // 在call()方法内部，就可以对原始RDD中的每一个元素进行各种处理和计算，并返回一个新的元素  
    // 所有新的元素就会组成一个新的RDD

    // 所有新的元素就会组成一个新的RDD  
    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 \* ;

        }  
    });

     multipleNumberRDD.foreach(new VoidFunction<Integer>() {

        private static final long serialVersionUID = 1L;

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

}

}

运行代码

join案例

打印学生成绩

package com.it19gong.sparkproject;

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.VoidFunction;

import scala.Tuple2;

public class TransforDemo {

public static void main(String\[\] args) {  
    //map();  
    //filter();  
    //flatMap() ;  
    // groupByKey();  
    //sortByKey();  
    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>(, "leo"),  
                    new Tuple2<Integer, String>(, "jack"),  
                    new Tuple2<Integer, String>(, "tom"));

            List<Tuple2<Integer, Integer>> scoreList = Arrays.asList(  
                    new Tuple2<Integer, Integer>(, ),  
                    new Tuple2<Integer, Integer>(, ),  
                    new Tuple2<Integer, Integer>(, ));

            // 并行化两个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的类型  
            // join，就返回的RDD的每一个元素，就是通过key join上的一个pair  
            // 什么意思呢？比如有(1, 1) (1, 2) (1, 3)的一个RDD  
                // 还有一个(1, 4) (2, 1) (2, 2)的一个RDD  
                // join以后，实际上会得到(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("===============================");  
                        }

                    });

            // 关闭JavaSparkContext  
            sc.close();

}

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>(, "leo"),  
            new Tuple2<Integer, String>(, "tom"),  
            new Tuple2<Integer, String>(, "marry"),  
            new Tuple2<Integer, String>(, "jack"));

    // 并行化集合，创建RDD  
    JavaPairRDD<Integer, String> scores = sc.parallelizePairs(scoreList);

    // 对scores RDD执行sortByKey算子  
    // sortByKey其实就是根据key进行排序，可以手动指定升序，或者降序  
    // 返回的，还是JavaPairRDD，其中的元素内容，都是和原始的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);  
        }

    });

    // 关闭JavaSparkContext  
    sc.close();

}

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", ),  
                    new Tuple2<String, Integer>("class2", ),  
                    new Tuple2<String, Integer>("class1", ),  
                    new Tuple2<String, Integer>("class2", ));

            // 并行化集合，创建JavaPairRDD  
            JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);

            // 针对scores RDD，执行groupByKey算子，对每个班级的成绩进行分组  
            // groupByKey算子，返回的还是JavaPairRDD  
            // 但是，JavaPairRDD的第一个泛型类型不变，第二个泛型类型变成Iterable这种集合类型  
            // 也就是说，按照了key进行分组，那么每个key可能都会有多个value，此时多个value聚合成了Iterable  
            // 那么接下来，我们是不是就可以通过groupedScores这种JavaPairRDD，很方便地处理某个分组内的数据  
            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();

}

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算子，将每一行文本，拆分为多个单词  
            // flatMap算子，在java中，接收的参数是FlatMapFunction  
            // 我们需要自己定义FlatMapFunction的第二个泛型类型，即，代表了返回的新元素的类型  
            // call()方法，返回的类型，不是U，而是Iterable<U>，这里的U也与第二个泛型类型相同  
            // flatMap其实就是，接收原始RDD中的每个元素，并进行各种逻辑的计算和处理，可以返回多个元素  
            // 多个元素，即封装在Iterable集合中，可以使用ArrayList等集合  
            // 新的RDD中，即封装了所有的新元素；也就是说，新的RDD的大小一定是 >= 原始RDD的大小  
            JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

                private static final long serialVersionUID = 1L;

                // 在这里会，比如，传入第一行，hello you  
                // 返回的是一个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);  
                }  
            });

            // 关闭JavaSparkContext  
            sc.close();

}

private static void filter() {  
            // 创建SparkConf  
            SparkConf conf = new SparkConf().setAppName("filter").setMaster("local");  
            // 创建JavaSparkContext  
            JavaSparkContext sc = new JavaSparkContext(conf);

            // 模拟集合  
            List<Integer> numbers = Arrays.asList(, , , , , , , , , );

            // 并行化集合，创建初始RDD  
            JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

            // 对初始RDD执行filter算子，过滤出其中的偶数  
            // filter算子，传入的也是Function，其他的使用注意点，实际上和map是一样的  
            // 但是，唯一的不同，就是call()方法的返回类型是Boolean  
            // 每一个初始RDD中的元素，都会传入call()方法，此时你可以执行各种自定义的计算逻辑  
            // 来判断这个元素是否是你想要的  
            // 如果你想在新的RDD中保留这个元素，那么就返回true；否则，不想保留这个元素，返回false

            JavaRDD<Integer> evenNumberRDD = numberRDD.filter(

                    new Function<Integer, Boolean>() {

                        private static final long serialVersionUID = 1L;

                        // 在这里，1到10，都会传入进来  
                        // 但是根据我们的逻辑，只有2,4,6,8,10这几个偶数，会返回true  
                        // 所以，只有偶数会保留下来，放在新的RDD中  
                        @Override  
                        public Boolean call(Integer v1) throws Exception {  
                            return v1 %  == ;  
                        }

                    });

            // 打印新的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();  
        }

private static void map() {  
    //创建SparkConf  
    SparkConf conf = new SparkConf().setAppName("map").setMaster("local");  
    // 创建JavaSparkContext  
    JavaSparkContext sc = new JavaSparkContext(conf);  
    //构建集合  
    List numbers = Arrays.asList(,,,,);  
    //并行化集合，创建初始RDD  
    JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

       // 使用map算子，将集合中的每个元素都乘以2  
    // map算子，是对任何类型的RDD，都可以调用的  
    // 在java中，map算子接收的参数是Function对象  
    // 创建的Function对象，一定会让你设置第二个泛型参数，这个泛型类型，就是返回的新元素的类型  
    // 同时call()方法的返回类型，也必须与第二个泛型类型同步  
    // 在call()方法内部，就可以对原始RDD中的每一个元素进行各种处理和计算，并返回一个新的元素  
    // 所有新的元素就会组成一个新的RDD

    // 所有新的元素就会组成一个新的RDD  
    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 \* ;

        }  
    });

     multipleNumberRDD.foreach(new VoidFunction<Integer>() {

        private static final long serialVersionUID = 1L;

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

}

}

运行代码

cogroup案例：打印学生成绩

package com.it19gong.sparkproject;

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.VoidFunction;

import scala.Tuple2;

public class TransforDemo {

public static void main(String\[\] args) {  
    //map();  
    //filter();  
    //flatMap() ;  
    // groupByKey();  
    //sortByKey();  
    //join();  
    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>(, "leo"),  
            new Tuple2<Integer, String>(, "jack"),  
            new Tuple2<Integer, String>(, "tom"));

    List<Tuple2<Integer, Integer>> scoreList = Arrays.asList(  
            new Tuple2<Integer, Integer>(, ),  
            new Tuple2<Integer, Integer>(, ),  
            new Tuple2<Integer, Integer>(, ),  
            new Tuple2<Integer, Integer>(, ),  
            new Tuple2<Integer, Integer>(, ),  
            new Tuple2<Integer, Integer>(, ));

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

    // cogroup与join不同  
    // 相当于是，一个key join上的所有value，都给放到一个Iterable里面去了  
    // cogroup，不太好讲解，希望大家通过动手编写我们的案例，仔细体会其中的奥妙  
    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();  
}

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>(, "leo"),  
                    new Tuple2<Integer, String>(, "jack"),  
                    new Tuple2<Integer, String>(, "tom"));

            List<Tuple2<Integer, Integer>> scoreList = Arrays.asList(  
                    new Tuple2<Integer, Integer>(, ),  
                    new Tuple2<Integer, Integer>(, ),  
                    new Tuple2<Integer, Integer>(, ));

            // 并行化两个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的类型  
            // join，就返回的RDD的每一个元素，就是通过key join上的一个pair  
            // 什么意思呢？比如有(1, 1) (1, 2) (1, 3)的一个RDD  
                // 还有一个(1, 4) (2, 1) (2, 2)的一个RDD  
                // join以后，实际上会得到(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("===============================");  
                        }

                    });

            // 关闭JavaSparkContext  
            sc.close();

}

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>(, "leo"),  
            new Tuple2<Integer, String>(, "tom"),  
            new Tuple2<Integer, String>(, "marry"),  
            new Tuple2<Integer, String>(, "jack"));

    // 并行化集合，创建RDD  
    JavaPairRDD<Integer, String> scores = sc.parallelizePairs(scoreList);

    // 对scores RDD执行sortByKey算子  
    // sortByKey其实就是根据key进行排序，可以手动指定升序，或者降序  
    // 返回的，还是JavaPairRDD，其中的元素内容，都是和原始的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);  
        }

    });

    // 关闭JavaSparkContext  
    sc.close();

}

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", ),  
                    new Tuple2<String, Integer>("class2", ),  
                    new Tuple2<String, Integer>("class1", ),  
                    new Tuple2<String, Integer>("class2", ));

            // 并行化集合，创建JavaPairRDD  
            JavaPairRDD<String, Integer> scores = sc.parallelizePairs(scoreList);

            // 针对scores RDD，执行groupByKey算子，对每个班级的成绩进行分组  
            // groupByKey算子，返回的还是JavaPairRDD  
            // 但是，JavaPairRDD的第一个泛型类型不变，第二个泛型类型变成Iterable这种集合类型  
            // 也就是说，按照了key进行分组，那么每个key可能都会有多个value，此时多个value聚合成了Iterable  
            // 那么接下来，我们是不是就可以通过groupedScores这种JavaPairRDD，很方便地处理某个分组内的数据  
            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();

}

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算子，将每一行文本，拆分为多个单词  
            // flatMap算子，在java中，接收的参数是FlatMapFunction  
            // 我们需要自己定义FlatMapFunction的第二个泛型类型，即，代表了返回的新元素的类型  
            // call()方法，返回的类型，不是U，而是Iterable<U>，这里的U也与第二个泛型类型相同  
            // flatMap其实就是，接收原始RDD中的每个元素，并进行各种逻辑的计算和处理，可以返回多个元素  
            // 多个元素，即封装在Iterable集合中，可以使用ArrayList等集合  
            // 新的RDD中，即封装了所有的新元素；也就是说，新的RDD的大小一定是 >= 原始RDD的大小  
            JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

                private static final long serialVersionUID = 1L;

                // 在这里会，比如，传入第一行，hello you  
                // 返回的是一个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);  
                }  
            });

            // 关闭JavaSparkContext  
            sc.close();

}

private static void filter() {  
            // 创建SparkConf  
            SparkConf conf = new SparkConf().setAppName("filter").setMaster("local");  
            // 创建JavaSparkContext  
            JavaSparkContext sc = new JavaSparkContext(conf);

            // 模拟集合  
            List<Integer> numbers = Arrays.asList(, , , , , , , , , );

            // 并行化集合，创建初始RDD  
            JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

            // 对初始RDD执行filter算子，过滤出其中的偶数  
            // filter算子，传入的也是Function，其他的使用注意点，实际上和map是一样的  
            // 但是，唯一的不同，就是call()方法的返回类型是Boolean  
            // 每一个初始RDD中的元素，都会传入call()方法，此时你可以执行各种自定义的计算逻辑  
            // 来判断这个元素是否是你想要的  
            // 如果你想在新的RDD中保留这个元素，那么就返回true；否则，不想保留这个元素，返回false

            JavaRDD<Integer> evenNumberRDD = numberRDD.filter(

                    new Function<Integer, Boolean>() {

                        private static final long serialVersionUID = 1L;

                        // 在这里，1到10，都会传入进来  
                        // 但是根据我们的逻辑，只有2,4,6,8,10这几个偶数，会返回true  
                        // 所以，只有偶数会保留下来，放在新的RDD中  
                        @Override  
                        public Boolean call(Integer v1) throws Exception {  
                            return v1 %  == ;  
                        }

                    });

            // 打印新的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();  
        }

private static void map() {  
    //创建SparkConf  
    SparkConf conf = new SparkConf().setAppName("map").setMaster("local");  
    // 创建JavaSparkContext  
    JavaSparkContext sc = new JavaSparkContext(conf);  
    //构建集合  
    List numbers = Arrays.asList(,,,,);  
    //并行化集合，创建初始RDD  
    JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

       // 使用map算子，将集合中的每个元素都乘以2  
    // map算子，是对任何类型的RDD，都可以调用的  
    // 在java中，map算子接收的参数是Function对象  
    // 创建的Function对象，一定会让你设置第二个泛型参数，这个泛型类型，就是返回的新元素的类型  
    // 同时call()方法的返回类型，也必须与第二个泛型类型同步  
    // 在call()方法内部，就可以对原始RDD中的每一个元素进行各种处理和计算，并返回一个新的元素  
    // 所有新的元素就会组成一个新的RDD

    // 所有新的元素就会组成一个新的RDD  
    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 \* ;

        }  
    });

     multipleNumberRDD.foreach(new VoidFunction<Integer>() {

        private static final long serialVersionUID = 1L;

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

}

}

运行代码

手机扫一扫

移动阅读更方便

阿里云服务器
腾讯云服务器
七牛云服务器

你可能感兴趣的文章