Spark核心编程-基础代码

使用Java，Scala和spark-shell开发wordcount程序

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>groupId</groupId>
    <artifactId>TestES</artifactId>
    <version>1.0-SNAPSHOT</version>
    <name>es-core-first</name>
    <url>http://maven.apache.org</url>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>transport</artifactId>
            <version>6.7.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
            <version>2.7</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.7</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>1.2.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.10</artifactId>
            <version>1.6.0</version>
        </dependency>
    </dependencies>
</project>

Java版

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.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
import java.util.Arrays;
public class WordCountLocal {
    public static void main(String[] args) {
        // 编写Spark应用程序
        // 本地执行，是可以执行在eclipse中的main方法中，执行的
        // 第一步：创建SparkConf对象，设置Spark应用的配置信息
        // 使用setMaster()可以设置Spark应用程序要连接的Spark集群的master节点的url
        // 但是如果设置为local则代表，在本地运行
        SparkConf conf = new SparkConf()
                .setAppName("WordCountLocal")
                .setMaster("local");
        // 第二步：创建JavaSparkContext对象
        // 在Spark中，SparkContext是Spark所有功能的一个入口，你无论是用java、scala，甚至是python编写
        // 都必须要有一个SparkContext，它的主要作用，包括初始化Spark应用程序所需的一些核心组件，包括
        // 调度器（DAGSchedule、TaskScheduler），还会去到Spark Master节点上进行注册，等等
        // 一句话，SparkContext，是Spark应用中，可以说是最最重要的一个对象
        // 但是呢，在Spark中，编写不同类型的Spark应用程序，使用的SparkContext是不同的，如果使用scala，
        // 使用的就是原生的SparkContext对象
        // 但是如果使用Java，那么就是JavaSparkContext对象
        // 如果是开发Spark SQL程序，那么就是SQLContext、HiveContext
        // 如果是开发Spark Streaming程序，那么就是它独有的SparkContext
        // 以此类推
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 第三步：要针对输入源（hdfs文件、本地文件，等等），创建一个初始的RDD
        // 输入源中的数据会打散，分配到RDD的每个partition中，从而形成一个初始的分布式的数据集
        // 我们这里呢，因为是本地测试，所以呢，就是针对本地文件
        // SparkContext中，用于根据文件类型的输入源创建RDD的方法，叫做textFile()方法
        // 在Java中，创建的普通RDD，都叫做JavaRDD
        // 在这里呢，RDD中，有元素这种概念，如果是hdfs或者本地文件呢，创建的RDD，每一个元素就相当于
        // 是文件里的一行
        JavaRDD<String> lines = sc.textFile("/Users/gaozhen/tmp/spark.txt");
        // 第四步：对初始RDD进行transformation操作，也就是一些计算操作
        // 通常操作会通过创建function，并配合RDD的map、flatMap等算子来执行
        // function，通常，如果比较简单，则创建指定Function的匿名内部类
        // 但是如果function比较复杂，则会单独创建一个类，作为实现这个function接口的类
        // 先将每一行拆分成单个的单词
        // FlatMapFunction，有两个泛型参数，分别代表了输入和输出类型
        // 我们这里呢，输入肯定是String，因为是一行一行的文本，输出，其实也是String，因为是每一行的文本
        // 这里先简要介绍flatMap算子的作用，其实就是，将RDD的一个元素，给拆分成一个或多个元素
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            private static final long serialVersionUID = 1L;
//            @Override
            public Iterable<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" "));
            }
        });
        // 接着，需要将每一个单词，映射为(单词, 1)的这种格式
        // 因为只有这样，后面才能根据单词作为key，来进行每个单词的出现次数的累加
        // mapToPair，其实就是将每个元素，映射为一个(v1,v2)这样的Tuple2类型的元素
        // 如果大家还记得scala里面讲的tuple，那么没错，这里的tuple2就是scala类型，包含了两个值
        // mapToPair这个算子，要求的是与PairFunction配合使用，第一个泛型参数代表了输入类型
        // 第二个和第三个泛型参数，代表的输出的Tuple2的第一个值和第二个值的类型
        // JavaPairRDD的两个泛型参数，分别代表了tuple元素的第一个值和第二个值的类型
        JavaPairRDD<String, Integer> pairs = words.mapToPair(
                new PairFunction<String, String, Integer>() {
                    private static final long serialVersionUID = 1L;
//                    @Override
                    public Tuple2<String, Integer> call(String word) throws Exception {
                        return new Tuple2<String, Integer>(word, 1);
                    }
                });
        // 接着，需要以单词作为key，统计每个单词出现的次数
        // 这里要使用reduceByKey这个算子，对每个key对应的value，都进行reduce操作
        // 比如JavaPairRDD中有几个元素，分别为(hello, 1) (hello, 1) (hello, 1) (world, 1)
        // reduce操作，相当于是把第一个值和第二个值进行计算，然后再将结果与第三个值进行计算
        // 比如这里的hello，那么就相当于是，首先是1 + 1 = 2，然后再将2 + 1 = 3
        // 最后返回的JavaPairRDD中的元素，也是tuple，但是第一个值就是每个key，第二个值就是key的value
        // reduce之后的结果，相当于就是每个单词出现的次数
        JavaPairRDD<String, Integer> wordCounts = pairs.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;
                    }
                });
        // 到这里为止，我们通过几个Spark算子操作，已经统计出了单词的次数
        // 但是，之前我们使用的flatMap、mapToPair、reduceByKey这种操作，都叫做transformation操作
        // 一个Spark应用中，光是有transformation操作，是不行的，是不会执行的，必须要有一种叫做action
        // 接着，最后，可以使用一种叫做action操作的，比如说，foreach，来触发程序的执行
        wordCounts.foreach(new VoidFunction<Tuple2<String,Integer>>() {
            private static final long serialVersionUID = 1L;
//            @Override
            public void call(Tuple2<String, Integer> wordCount) throws Exception {
                System.out.println(wordCount._1 + " appeared " + wordCount._2 + " times.");
            }
        });
        sc.close();
    }
}

scala版本：

import org.apache.spark.{SparkConf, SparkContext}
object WordCount {
    def main(args: Array[String]) {
      val conf = new SparkConf()
        .setAppName("WordCount").setMaster("local");
      val sc = new SparkContext(conf)
      val lines = sc.textFile("/Users/gaozhen/tmp/spark.txt", 1);
      val words = lines.flatMap { line => line.split(" ") }
      val pairs = words.map { word => (word, 1) }
      val wordCounts = pairs.reduceByKey { _ + _ }
      wordCounts.foreach(wordCount => println(wordCount._1 + " appeared " + wordCount._2 + " times."))
    }
}

遇到的问题：

Exception in thread "main" java.lang.NoSuchMethodError: scala.Predef$.refArrayOps([Ljava/lang/Object;)Lscala/collection/mutable/ArrayOps;
原因： 使用的spark版本是1.6的，spark自身带的Scala版本是2.10.* 出现spark的Scala版本不一致导致的
解决方案：
将Scala版本降到2.10.*版本

Spark核心编程：transformation操作开发实战

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

Java版

   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
        // map算子，是对任何类型的RDD，都可以调用的
        // 在java中，map算子接收的参数是Function对象
        // 创建的Function对象，一定会让你设置第二个泛型参数，这个泛型类型，就是返回的新元素的类型
            // 同时call()方法的返回类型，也必须与第二个泛型类型同步
        // 在call()方法内部，就可以对原始RDD中的每一个元素进行各种处理和计算，并返回一个新的元素
        // 所有新的元素就会组成一个新的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);  
            }
        });
        // 关闭JavaSparkContext
        sc.close();
    }

Scala版

  def map(): Unit = {
    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))
  }

filter：过滤出集合中的偶数

java版本

    /**
     * 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算子，过滤出其中的偶数
        // 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 % 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();
    }

scala 版本

  def filter(): Unit = {
    val conf = new SparkConf().setAppName("filter").setMaster("local");
    val sc = new SparkContext(conf);
    val numbers = Array(1,2,3,4,5,6,7,8);
    val numberRdd = sc.parallelize(numbers,1);
    val evenNumber = numberRdd.filter{num => num % 2 == 0}
    evenNumber.foreach(num => println(num))
  }

flatMap：将行拆分为单词

java版本

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

scala 版本

  def flatMap(): Unit = {
    val conf = new SparkConf();
    conf.setAppName("flatMap");
    conf.setMaster("local");
    val sc = new SparkContext(conf);
    val lineArray = Array("hello you", "cbd dfe sfewfw");
    val lines = sc.parallelize(lineArray);
    val words = lines.flatMap(line=>line.split(" "));
    words.foreach(word=>println(word))
  }

groupByKey案例：按照班级对成绩进行分组

java 版本

    /**
     * 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算子，对每个班级的成绩进行分组
        // 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();
    }

scala 版本

  def groupByKey(): Unit ={
    val conf = new SparkConf().setAppName("groupByKey").setMaster("local");
    val sc = new SparkContext(conf);
    val scoreList = Array(Tuple2("class1",80), Tuple2("class2", 75), Tuple2("class1",40))
    val scores = sc.parallelize(scoreList, 1)
    val groupedScores = scores.groupByKey()
    groupedScores.foreach(score => {
      println(score._1)
      score._2.foreach( singleScore => println(singleScore))
      println("=============")
    })
  }

reduceByKey案例：统计每个班级的总分

Java版本

    /**
     * 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算子
        // reduceByKey，接收的参数是Function2类型，它有三个泛型参数，实际上代表了三个值
        // 第一个泛型类型和第二个泛型类型，代表了原始RDD中的元素的value的类型
            // 因此对每个key进行reduce，都会依次将第一个、第二个value传入，将值再与第三个value传入
            // 因此此处，会自动定义两个泛型类型，代表call()方法的两个传入参数的类型
        // 第三个泛型类型，代表了每次reduce操作返回的值的类型，默认也是与原始RDD的value类型相同的
        // reduceByKey算法返回的RDD，还是JavaPairRDD<key, value>
        JavaPairRDD<String, Integer> totalScores = scores.reduceByKey(
                new Function2<Integer, Integer, Integer>() {
                    private static final long serialVersionUID = 1L;
                    // 对每个key，都会将其value，依次传入call方法
                    // 从而聚合出每个key对应的一个value
                    // 然后，将每个key对应的一个value，组合成一个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);   
            }
        });
        // 关闭JavaSparkContext
        sc.close();
    }

scala 版本

  def reduceByKey(): Unit = {
    val cof = new SparkConf().setAppName("recudeByKey").setMaster("local");
    val sc = new SparkContext(cof);
    val scoreList = Array(Tuple2("class1", 80), Tuple2("class2", 54),
      Tuple2("class2", 98))
    val scores = sc.parallelize(scoreList,1)
    val totalScores = scores.reduceByKey(_+_)
    totalScores.foreach(classScore => println(classScore._1+ ":" + classScore._2))
  }

sortByKey案例：按照学生分数进行排序
java 版本

    /**
     * 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算子
        // 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();
    }

Scala版本

  def sortByKey(): Unit = {
    val conf = new SparkConf().setAppName("sortByKey").setMaster("local");
    val sc = new SparkContext(conf)
    val scoresList = Array(Tuple2(54,"tome"), Tuple2(76, "tome2"));
    val socres = sc.parallelize(scoresList, 1)
    val sortedScores = socres.sortByKey(false);
    sortedScores.foreach(studentScore => println(studentScore._1 + ":" + studentScore._2))
  }

join案例：打印学生成绩

java 版本

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

Scala版本

 def join(): Unit = {
    val conf = new SparkConf();
    conf.setAppName("join");
    conf.setMaster("local");
    val sc = new SparkContext(conf);
    val studentList = Array(
      Tuple2(1,"leo"),
      Tuple2(2,"jack"),
      Tuple2(3,"tom"))
    val scoreList = Array(
      Tuple2(1,300),
      Tuple2(2,432),
      Tuple2(3,329));
    val students = sc.parallelize(studentList);
    val scores = sc.parallelize(scoreList);
    val studentScores = students.join(scores);
    studentScores.foreach(studentScores => {
      println("student id: " + studentScores._1);
      println("student name:" + studentScores._2._1);
      println("student score:" + studentScores._2._1);
    })
  }

cogroup案例：打印学生成绩

java 版本

    /**
     * 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里面去了 
        // 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();
    }

action操作开发实战

reduce
java 版本

    private static void reduce() {
        // 创建SparkConf和JavaSparkContext
        SparkConf conf = new SparkConf()
                .setAppName("reduce")
                .setMaster("local");  
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 有一个集合，里面有1到10,10个数字，现在要对10个数字进行累加
        List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD<Integer> numbers = sc.parallelize(numberList);
        // 使用reduce操作对集合中的数字进行累加
        // reduce操作的原理：
            // 首先将第一个和第二个元素，传入call()方法，进行计算，会获取一个结果，比如1 + 2 = 3
            // 接着将该结果与下一个元素传入call()方法，进行计算，比如3 + 3 = 6
            // 以此类推
        // 所以reduce操作的本质，就是聚合，将多个元素聚合成一个元素
        int sum = numbers.reduce(new Function2<Integer, Integer, Integer>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        System.out.println(sum);  
        // 关闭JavaSparkContext
        sc.close();
    }

Scala版本

  def reduce(): Unit = {
    val conf = new SparkConf().setAppName("reduce").setMaster("local");
    val sc = new SparkContext(conf);
    val numberArray = Array(1,2,3,4,5,6,7,8)
    val numbers = sc.parallelize(numberArray, 1)
    val sum = numbers.reduce(_+_)
    println(sum)
  }

collect
java 版本

    private static void collect() {
        // 创建SparkConf和JavaSparkContext
        SparkConf conf = new SparkConf()
                .setAppName("collect")
                .setMaster("local");  
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 有一个集合，里面有1到10,10个数字，现在要对10个数字进行累加
        List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD<Integer> numbers = sc.parallelize(numberList);
        // 使用map操作将集合中所有数字乘以2
        JavaRDD<Integer> doubleNumbers = numbers.map(
                new Function<Integer, Integer>() {
                    private static final long serialVersionUID = 1L;
                    @Override
                    public Integer call(Integer v1) throws Exception {
                        return v1 * 2;
                    }
                });
        // 不用foreach action操作，在远程集群上遍历rdd中的元素
        // 而使用collect操作，将分布在远程集群上的doubleNumbers RDD的数据拉取到本地
        // 这种方式，一般不建议使用，因为如果rdd中的数据量比较大的话，比如超过1万条
            // 那么性能会比较差，因为要从远程走大量的网络传输，将数据获取到本地
            // 此外，除了性能差，还可能在rdd中数据量特别大的情况下，发生oom异常，内存溢出
        // 因此，通常，还是推荐使用foreach action操作，来对最终的rdd元素进行处理
        List<Integer> doubleNumberList = doubleNumbers.collect();
        for(Integer num : doubleNumberList) {
            System.out.println(num);  
        }
        // 关闭JavaSparkContext
        sc.close();
    }

scala版本

  def collect(): Unit = {
    val conf = new SparkConf().setAppName("collect").setMaster("local");
    val sc = new SparkContext(conf);
    val numberArray = Array(1,2,3,4,5,6,7,8,9);
    val numbers = sc.parallelize(numberArray, 1)
    val sum = numbers.reduce(_ + _)
    println(sum)
  }

count
java 版本

    private static void count() {
        // 创建SparkConf和JavaSparkContext
        SparkConf conf = new SparkConf()
                .setAppName("count")
                .setMaster("local");  
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 有一个集合，里面有1到10,10个数字，现在要对10个数字进行累加
        List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD<Integer> numbers = sc.parallelize(numberList);
        // 对rdd使用count操作，统计它有多少个元素
        long count = numbers.count();
        System.out.println(count);  
        // 关闭JavaSparkContext
        sc.close();
    }

scala版本

  def count(): Unit = {
    val conf = new SparkConf().setAppName("count").setMaster("local");
    val sc = new SparkContext(conf);
     val numberArray = Array(1,2,3,4,5,6,7,8,9);
    val numbers = sc.parallelize(numberArray);
    val count = numbers.count();
    println(count)
  }

take
java版本

    private static void take() {
        // 创建SparkConf和JavaSparkContext
        SparkConf conf = new SparkConf()
                .setAppName("take")
                .setMaster("local");  
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 有一个集合，里面有1到10,10个数字，现在要对10个数字进行累加
        List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD<Integer> numbers = sc.parallelize(numberList);
        // 对rdd使用count操作，统计它有多少个元素
        // take操作，与collect类似，也是从远程集群上，获取rdd的数据
        // 但是collect是获取rdd的所有数据，take只是获取前n个数据
        List<Integer> top3Numbers = numbers.take(3);
        for(Integer num : top3Numbers) {
            System.out.println(num);  
        }
        // 关闭JavaSparkContext
        sc.close();
    }

scala 版本

  def take(): Unit = {
    val conf = new SparkConf().setAppName("take").setMaster("local");
    val sc = new SparkContext(conf);
    val numberArray = Array(1,2,3,4,5,6,7);
    val numbers = sc.parallelize(numberArray);
    val top3Number = numbers.take(3);
    for(num <- top3Number) {
      println(num)
    }
  }

saveAsTextFile
Java版本

    private static void saveAsTextFile() {
        // 创建SparkConf和JavaSparkContext
        SparkConf conf = new SparkConf()
                .setAppName("saveAsTextFile");  
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 有一个集合，里面有1到10,10个数字，现在要对10个数字进行累加
        List<Integer> numberList = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        JavaRDD<Integer> numbers = sc.parallelize(numberList);
        // 使用map操作将集合中所有数字乘以2
        JavaRDD<Integer> doubleNumbers = numbers.map(
                new Function<Integer, Integer>() {
                    private static final long serialVersionUID = 1L;
                    @Override
                    public Integer call(Integer v1) throws Exception {
                        return v1 * 2;
                    }
                });
        // 直接将rdd中的数据，保存在HFDS文件中
        // 但是要注意，我们这里只能指定文件夹，也就是目录
        // 那么实际上，会保存为目录中的/double_number.txt/part-00000文件
        doubleNumbers.saveAsTextFile("hdfs://spark1:9000/double_number.txt");   
        // 关闭JavaSparkContext
        sc.close();
    }

countByKey
java版本

    private static void countByKey() {
        // 创建SparkConf
        SparkConf conf = new SparkConf()
                .setAppName("countByKey")  
                .setMaster("local");
        // 创建JavaSparkContext
        JavaSparkContext sc = new JavaSparkContext(conf);
        // 模拟集合
        List<Tuple2<String, String>> scoreList = Arrays.asList(
                new Tuple2<String, String>("class1", "leo"),
                new Tuple2<String, String>("class2", "jack"),
                new Tuple2<String, String>("class1", "marry"),
                new Tuple2<String, String>("class2", "tom"),
                new Tuple2<String, String>("class2", "david"));  
        // 并行化集合，创建JavaPairRDD
        JavaPairRDD<String, String> students = sc.parallelizePairs(scoreList);
        // 对rdd应用countByKey操作，统计每个班级的学生人数，也就是统计每个key对应的元素个数
        // 这就是countByKey的作用
        // countByKey返回的类型，直接就是Map<String, Object>
        Map<String, Object> studentCounts = students.countByKey();
        for(Map.Entry<String, Object> studentCount : studentCounts.entrySet()) {
            System.out.println(studentCount.getKey() + ": " + studentCount.getValue());  
        }
        // 关闭JavaSparkContext
        sc.close();
    }

scala版本

  def countByKey(): Unit = {
    val conf = new SparkConf().setAppName("countByKey").setMaster("local");
    val sc = new SparkContext(conf);
    val studentList = Array(Tuple2("class1","leo"), Tuple2("class2","tom"), Tuple2("class4", "adbc"));
    val students = sc.parallelize(studentList);
    val studentCounts = students.countByKey();
    println(studentCounts)
  }