5、创建RDD（集合、本地文件、HDFS文件）

一、创建RDD

1、创建RDD

进行Spark核心编程时，首先要做的第一件事，就是创建一个初始的RDD。该RDD中，通常就代表和包含了Spark应用程序的输入源数据。然后在创建了初始的RDD之后，才可以通过Spark Core提供的transformation算子，对该RDD进行转换，来获取其他的RDD。

Spark Core提供了三种创建RDD的方式，包括：使用程序中的集合创建RDD；使用本地文件创建RDD；使用HDFS文件创建RDD。

1、使用程序中的集合创建RDD，主要用于进行测试，可以在实际部署到集群运行之前，自己使用集合构造测试数据，来测试后面的spark应用的流程。

2、使用本地文件创RDD，主要用于临时性地处理一些存储了大量数据的文件。

3、使用HDFS文件创建RDD，应该是最常用的生产环境处理方式，主要可以针对HDFS上存储的大数据，进行离线批处理操作。

2、并行化集合创建RDD

如果要通过并行化集合来创建RDD，需要针对程序中的集合，调用SparkContext的parallelize()方法。Spark会将集合中的数据拷贝到集群上去，形成一个分布式的数据集合，也就是一个RDD。相当于是，集合中的部分数据会到一个节点上，而另一部分数据会到其他节点上。然后就可以用并行的方式来操作这个分布式数据集合，即RDD。

调用parallelize()时，有一个重要的参数可以指定，就是要将集合切分成多少个partition。Spark会为每一个partition运行一个task来进行处理。Spark官方的建议是，为集群中

的每个CPU创建2~4个partition。Spark默认会根据集群的情况来设置partition的数量。但是也可以在调用parallelize()方法时，传入第二个参数，来设置RDD的partition数量。

比如parallelize(arr, 10)

// 案例：1到10累加求和

###java实现

package cn.spark.study.core;

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

/**

 * 并行化集合创建RDD

 *

 * @author bcqf

 *

 */

public class ParallelizeCollection {

    public static void main(String[] args) {

        // 创建SparkConf

        SparkConf conf = new SparkConf().setAppName("ParallelizeCollection").setMaster("local");

        // 创建JavaSparkContext

        JavaSparkContext sc = new JavaSparkContext(conf);

        // 要通过并行化集合的方式创建RDD，那么就调用SparkContext以及其子类，的parallelize()方法

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

        JavaRDD<Integer> numberRDD = sc.parallelize(numbers);

        // 执行reduce算子操作

                // 相当于，先进行1 + 2 = 3；然后再用3 + 3 = 6；然后再用6 + 4 = 10...以此类推;

        int sum = numberRDD.reduce(new Function2<Integer, Integer, Integer>() {

            private static final long serialVersionUID = 1L;

            @Override

            public Integer call(Integer num1, Integer num2) throws Exception {

                // TODO Auto-generated method stub

                return num1 + num2;

            }

        });

        //输入累加和

        System.out.println("1-10的累加和：" + sum);

        // 关闭JavaSparkContext

        sc.close();

    }

}

###scala实现

package cn.spark.study.core

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

object ParallelizeCollection {

  def main(args: Array[String]) {

    val conf = new SparkConf().setAppName("ParallelizeCollection").setMaster("local");

    val sc = new SparkContext(conf)

    val numbers = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

    val numberRDD = sc.parallelize(numbers, 5)

    val sum = numberRDD.reduce(_ + _)

    println("1-10的累加和：" + sum)

  }

}

3、使用本地文件和HDFS创建RDD

Spark是支持使用任何Hadoop支持的存储系统上的文件创建RDD的，比如说HDFS、Cassandra、HBase以及本地文件。通过调用SparkContext的textFile()方法，可以针对本地文件或HDFS文件创建RDD。

有几个事项是需要注意的：

1、如果是针对本地文件的话，如果是在windows上本地测试，windows上有一份文件即可；如果是在spark集群上针对linux本地文件，那么需要将文件拷贝到所有worker节点上。

2、Spark的textFile()方法支持针对目录、压缩文件以及通配符进行RDD创建。

3、Spark默认会为hdfs文件的每一个block创建一个partition，但是也可以通过textFile()的第二个参数手动设置分区数量，只能比block数量多，不能比block数量少。

-------------------案例：文件字数统计   ；本地文件-------------------

##java实现

package cn.spark.study.core;

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

/**

 * 使用本地文件创建RDD

 * 案例：统计文本文件字数

 * @author bcqf

 *

 */

public class LocalFile {

    public static void main(String[] args) {

        // 创建SparkConf

        SparkConf conf = new SparkConf().setAppName("LocalFile").setMaster("local");

        // 创建JavaSparkContext

        JavaSparkContext sc = new JavaSparkContext(conf);

        // 调用SparkContext以及其子类的textFile()方法，针对本地文件创建RDD

        JavaRDD<String> lines = sc.textFile("D://spark.txt");

        // 统计文本文件内的字数; Function<String, Integer> :String是接收类型，Integer是返回类型

        JavaRDD<Integer> lineLength = lines.map(new Function<String, Integer>() {

            private static final long serialVersionUID = 1L;

            @Override

            public Integer call(String v1) throws Exception {

                return v1.length();

            }

        });

        
        //Function2<T1, T2, R> ;  call(T1 v1, T2 v2)

        int count = lineLength.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("文字总数是：" + count);

        // 关闭JavaSparkContext

        sc.close();

    }

}

-------------------案例：文件字数统计   ；HDFS文件-------------------

##java实现

package cn.spark.study.core;

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

/**

 * 使用HDFS文件创建RDD

 * 案例：统计文本文件字数

 * @author bcqf

 *

 */

public class HDFSFile {

    public static void main(String[] args) {

        // 创建SparkConf

        SparkConf conf = new SparkConf().setAppName("HDFSFile");

        // 创建JavaSparkContext

        JavaSparkContext sc = new JavaSparkContext(conf);

        // 调用SparkContext以及其子类的textFile()方法，针对HDFS文件创建RDD

        JavaRDD<String> lines = sc.textFile("hdfs://spark1:9000/spark.txt");

        // 统计文本文件内的字数

        JavaRDD<Integer> lineLength = lines.map(new Function<String, Integer>() {

            private static final long serialVersionUID = 1L;

            @Override

            public Integer call(String v1) throws Exception {

                return v1.length();

            }

        });

        int count = lineLength.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("文字总数是：" + count);

        // 关闭JavaSparkContext

        sc.close();

    }

}

[root@spark1 java]# cat hdfs_file.sh         #运行脚本

/usr/local/spark/bin/spark-submit \

--class cn.spark.study.core.HDFSFile \

--num-executors 3 \

--driver-memory 100m \

--executor-memory 100m \

--executor-cores 3 \

/usr/local/spark-study/java/saprk-study-java-0.0.1-SNAPSHOT-jar-with-dependencies.jar \

##打maven包-->上传-->运行

-------------------案例：文件字数统计   ；本地文件-------------------

##scala实现

package cn.spark.study.core

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

object LocalFile {

  def main(args: Array[String]) {

    val conf = new SparkConf().setAppName("LocalFile").setMaster("local")

    val sc = new SparkContext(conf)

    val lines = sc.textFile("D://spark.txt", 1)

    val count = lines.map {line => line.length() }.reduce(_ + _)

    println("file's count is: " + count)

  }

}

-------------------案例：文件字数统计   ；HDFS文件-------------------

##scala实现

package cn.spark.study.core

import org.apache.spark.SparkConf

import org.apache.spark.SparkContext

object HDFSFile {

  def main(args: Array[String]) {

    val conf = new SparkConf().setAppName("HDFSFile")

    val sc = new SparkContext(conf)

    val lines = sc.textFile("hdfs://spark1:9000/spark.txt", 1)

    val count = lines.map {line => line.length() }.reduce(_ + _)

    println("file's count is: " + count)

  }

}

[root@spark1 scala]# cat hdfs_file.sh         #运行脚本

/usr/local/spark/bin/spark-submit \

--class cn.spark.study.core.HDFSFile \

--num-executors 3 \

--driver-memory 100m \

--executor-memory 100m \

--executor-cores 3 \

/usr/local/spark-study/scala/spark-study-scala.jar \

##打jar包(spark-study-scala.jar)-->上传-->运行