RDD：创建的几种方式（scala和java）[通俗易懂]

2022-06-29 09:02:10 浏览数 (1)

大家好，又见面了，我是你们的朋友全栈君。

Spark编程

每一个spark应用程序都包含一个驱动程序（driver program ），他会运行用户的main函数，并在集群上执行各种并行操作（parallel operations）

spark提供的最主要的抽象概念有两种： 弹性分布式数据集（resilient distributed dataset）简称RDD ，他是一个元素集合，被分区地分布到集群的不同节点上，可以被并行操作，RDD可以从hdfs(或者任意其他的支持Hadoop的文件系统)上的一个文件开始创建，或者通过转换驱动程序中已经存在的集合得到。

用户也可以让spark将一个RDD持久化到内存中，使其能再并行操作中被有效地重复使用，最后RDD能自动从节点故障中恢复。

spark的第二个抽象概念是共享变量（shared variables），它可以在并行操作中使用，在默认情况下，当spark将一个函数以任务集的形式在不同的节点上并行运行时，会将该函数所使用的每个变量拷贝传递给每一个任务中（因为每一个都需要拷贝，没必要，所以才设置共享变量），有时候，一个变量需要在任务之间，或者驱动程序之间进行共享。spark支持两种共享变量： 广播变量（broadcast variables），它可以在所有节点的内存中缓存一个值。 累加器（accumulators）：只能用于做加法的变量，例如计算器或求和器

RDD简介

  1. RDD是Spark的核心抽象，全称为Resillient Distributed Dataset，即弹性分布式数据集
  2. RDD在抽象上来说是一种元素集合，包含了数据。它是被分区的，分为多个分区，每个分区分布在集群中的不同节点上(分区即partition)，从而让RDD中的数据可以被并行操作。(分布式的特性)
  3. RDD通常通过Hadoop上的文件，即HDFS文件，来进行创建；有时也可以通过Spark应用程序中的集合来创建。
  4. RDD最重要的特性就是，提供了容错性，可以自动从节点失败中恢复过来。即如果某个节点上的RDD partition，因为节点故障，导致数据丢了，那么RDD会自动通过自己的数据来源重新计算partition。这一切对使用者是透明的。
  5. RDD的数据默认的情况下是存放在内存中的，但是在内存资源不足时，Spark会自动将RDD数据写入磁盘。(弹性的特性)

scala中创建RDD的三种方式

在RDD中，通常就代表和包含了Spark应用程序的输入源数据。 当我们，在创建了初始的RDD之后，才可以通过Spark Core提供的transformation算子，对该RDD进行transformation(转换)操作，来获取其他的RDD。 Spark Core为我们提供了三种创建RDD的方式，包括： 1. 使用程序中的集合创建RDD 2. 使用本地文件创建RDD 3. 使用HDFS文件创建RDD

应用场景 1. 使用程序中的集合创建RDD，主要用于进行测试，可以在实际部署到集群运行之前，自己使用集合构造测试数据，来测试后面的spark应用的流程 2. 使用本地文件创建RDD，主要用于的场景为：在本地临时性地处理一些存储了大量数据的文件 3. 使用HDFS文件创建RDD，应该是最常用的生产环境处理方式，主要可以针对HDFS上存储的大数据，进行离线批处理操作

实际操作

并行化创建RDD

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

代码语言：javascript复制
// 并行化创建RDD部分代码 
// 实现1到5的累加求和
val arr = Array(1,2,3,4,5)
val rdd = sc.parallelize(arr)
val sum = rdd.reduce(_   _)

官网的注意点 通过阅读Spark的官方文档，如下图：

我们可知： 在调用parallelize()方法时，有一个重要的参数可以指定，就是要将集合切分成多少个partition。Spark会为每一个partition运行一个task来进行处理。Spark官方的建议是，为集群中的每个CPU创建2-4个partition。Spark默认会根据集群的情况来设置partition的数量。但是也可以在调用parallelize()方法时，传入第二个参数，来设置RDD的partition数量。比如，parallelize(arr, 10)

使用textFile方法，通过本地文件或HDFS创建RDD

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

代码语言：javascript复制
// 实现文件字数统计
// textFile()方法中，输入本地文件路径或是HDFS路径
// HDFS:hdfs://spark1:9000/data.txt
// local:/home/hadoop/data.txt
val rdd = sc.textFile(“/home/hadoop/data.txt”)
val wordCount = rdd.map(line => line.length).reduce(_   _)

通过本地文件或HDFS创建RDD的几个注意点 1. 如果是针对本地文件的话： * 如果是在Windows上进行本地测试，windows上有一份文件即可； * 如果是在Spark集群上针对Linux本地文件，那么需要将文件拷贝到所有worker节点上（就是在spark-submit上使用—master指定了master节点，使用standlone模式进行运行，而textFile()方法内仍然使用的是Linux本地文件，在这种情况下，是需要将文件拷贝到所有worker节点上的）； 2. Spark的textFile()方法支持针对目录、压缩文件以及通配符进行RDD创建 3. Spark默认会为hdfs文件的每一个block创建一个partition，但是也可以通过textFile()的第二个参数手动设置分区数量，只能比block数量多，不能比block数量少

Spark支持的其余方法，创建RDD

同样，通过阅读Spark的官方文档，可以知道除了通过使用textFile()方法创建RDD之外，还有几个其余的方法适用于其它的应用场景，如下图：

SparkContext的textFile()除了可以针对上述几种普通的文件创建RDD之外，还有一些特例的方法来创建RDD：

  1. SparkContext的wholeTextFiles()方法，可以针对一个目录中的大量小文件，返回由（fileName,fileContent）组成的pair，即pairRDD，而不是普通的RDD。该方法返回的是文件名字和文件中的具体内容；而普通的textFile()方法返回的RDD中，每个元素就是文本中一行文本。
  2. SparkContext的sequenceFileK,V方法，可以针对SequenceFile创建RDD，K和V泛型类型就是SequenceFile的key和value的类型。K和V要求必须是Hadoop的序列化机制，比如IntWritable、Text等。
  3. SparkContext的hadoopRDD()方法，对于Hadoop的自定义输入类型，可以创建RDD。该方法接收JobConf、InputFormatClass、Key和Value的Class。
  4. SparkContext的objectFile()方法，可以针对之前调用的RDD的saveAsObjectFile()创建的对象序列化的文件，反序列化文件中的数据，并创建一个RDD。

参考原文：https://blog.csdn.net/lemonZhaoTao/article/details/77923337

java中创建RDD的两种方式

下面开始初始化spark

spark程序需要做的第一件事情，就是创建一个SparkContext对象，它将告诉spark如何访问一个集群，而要创建一个SparkContext对象，你首先要创建一个SparkConf对象，该对象访问了你的应用程序的信息

比如下面的代码是运行在spark模式下

代码语言：javascript复制
public class sparkTestCon {

    public static void main(String[] args) {
        SparkConf conf=new SparkConf();
        conf.set("spark.testing.memory", "2147480000");     //因为jvm无法获得足够的资源
        JavaSparkContext sc = new JavaSparkContext("spark://192.168.52.140:7077", "First Spark App",conf);
        System.out.println(sc);
    }

}

下面是运行在本机，把上面的第6行代码改为如下

代码语言：javascript复制
JavaSparkContext sc = new JavaSparkContext("local", "First Spark App",conf);

RDD的创建有两种方式 1.引用外部文件系统的数据集（HDFS） 2.并行化一个已经存在于驱动程序中的集合（并行集合，是通过对于驱动程序中的集合调用JavaSparkContext.parallelize来构建的RDD）

第一种方式创建 下面通过代码来理解RDD和怎么操作RDD

代码语言：javascript复制
package com.tg.spark;

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;
import org.apache.spark.storage.StorageLevel;
/**
 * 引用外部文件系统的数据集（HDFS）创建RDD
 *  匿名内部类定义函数传给spark
 * @author 汤高
 *
 */
public class RDDOps {
    //完成对所有行的长度求和
    public static void main(String[] args) {

        SparkConf conf=new SparkConf();
        conf.set("spark.testing.memory", "2147480000");     //因为jvm无法获得足够的资源
        JavaSparkContext sc = new JavaSparkContext("local", "First Spark App",conf);
        System.out.println(sc);

        //通过hdfs上的文件定义一个RDD 这个数据暂时还没有加载到内存，也没有在上面执行动作,lines仅仅指向这个文件
        JavaRDD<String> lines = sc.textFile("hdfs://master:9000/testFile/README.md");

        //定义lineLengths作为Map转换的结果 由于惰性，不会立即计算lineLengths
        //第一个参数为传入的内容，第二个参数为函数操作完后返回的结果类型
        JavaRDD<Integer> lineLengths = lines.map(new Function<String, Integer>() {
          public Integer call(String s) { 
              System.out.println("每行长度" s.length());
              return s.length(); }
        });
        //运行reduce  这是一个动作action  这时候，spark才将计算拆分成不同的task，
        //并运行在独立的机器上，每台机器运行他自己的map部分和本地的reducation，并返回结果集给去驱动程序
        int totalLength = lineLengths.reduce(new Function2<Integer, Integer, Integer>() {
          public Integer call(Integer a, Integer b) { return a   b; }
        });

        System.out.println(totalLength);
        //为了以后复用  持久化到内存...
        lineLengths.persist(StorageLevel.MEMORY_ONLY());


    }
}

如果觉得刚刚那种写法难以理解，可以看看第二种写法

代码语言：javascript复制
package com.tg.spark;
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;
import org.apache.spark.storage.StorageLevel;
/**
 * 引用外部文件系统的数据集（HDFS）创建RDD 
 *  外部类定义函数传给spark
 * @author 汤高
 *
 */
public class RDDOps2 {
    // 完成对所有行的长度求和
    public static void main(String[] args) {

        SparkConf conf = new SparkConf();
        conf.set("spark.testing.memory", "2147480000"); // 因为jvm无法获得足够的资源
        JavaSparkContext sc = new JavaSparkContext("local", "First Spark App", conf);
        System.out.println(sc);


        //通过hdfs上的文件定义一个RDD 这个数据暂时还没有加载到内存，也没有在上面执行动作,lines仅仅指向这个文件
        JavaRDD<String> lines = sc.textFile("hdfs://master:9000/testFile/README.md");
        //定义lineLengths作为Map转换的结果 由于惰性，不会立即计算lineLengths
        JavaRDD<Integer> lineLengths = lines.map(new GetLength());


        //运行reduce  这是一个动作action  这时候，spark才将计算拆分成不同的task，
                //并运行在独立的机器上，每台机器运行他自己的map部分和本地的reducation，并返回结果集给去驱动程序
        int totalLength = lineLengths.reduce(new Sum());

        System.out.println("总长度" totalLength);
        // 为了以后复用 持久化到内存...
        lineLengths.persist(StorageLevel.MEMORY_ONLY());

    }
    //定义map函数
    //第一个参数为传入的内容，第二个参数为函数操作完后返回的结果类型
    static class GetLength implements Function<String, Integer> {
        public Integer call(String s) {
            return s.length();
        }
    }
    //定义reduce函数 
    //第一个参数为内容，第三个参数为函数操作完后返回的结果类型
    static class Sum implements Function2<Integer, Integer, Integer> {
        public Integer call(Integer a, Integer b) {
            return a   b;
        }
    }
}

第二种方式创建RDD

代码语言：javascript复制
package com.tg.spark;

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.Function2;
import org.apache.spark.storage.StorageLevel;

import com.tg.spark.RDDOps2.GetLength;
import com.tg.spark.RDDOps2.Sum;
/**
 * 并行化一个已经存在于驱动程序中的集合创建RDD
 * @author 汤高
 *
 */
public class RDDOps3 {
    // 完成对所有数求和
    public static void main(String[] args) {

        SparkConf conf = new SparkConf();
        conf.set("spark.testing.memory", "2147480000"); // 因为jvm无法获得足够的资源
        JavaSparkContext sc = new JavaSparkContext("local", "First Spark App", conf);
        System.out.println(sc);

        List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
        //并行集合，是通过对于驱动程序中的集合调用JavaSparkContext.parallelize来构建的RDD
        JavaRDD<Integer> distData = sc.parallelize(data);

        JavaRDD<Integer> lineLengths = distData.map(new GetLength());

        // 运行reduce 这是一个动作action 这时候，spark才将计算拆分成不同的task，
        // 并运行在独立的机器上，每台机器运行他自己的map部分和本地的reducation，并返回结果集给去驱动程序
        int totalLength = lineLengths.reduce(new Sum());

        System.out.println("总和"   totalLength);
        // 为了以后复用 持久化到内存...
        lineLengths.persist(StorageLevel.MEMORY_ONLY());

    }

    // 定义map函数
    static class GetLength implements Function<Integer, Integer> {

        @Override
        public Integer call(Integer a) throws Exception {

            return a;
        }
    }

    // 定义reduce函数
    static class Sum implements Function2<Integer, Integer, Integer> {
        public Integer call(Integer a, Integer b) {
            return a   b;
        }
    }
}

注意：上面的写法是基于jdk1.7或者更低版本 基于jdk1.8有更简单的写法 下面是官方文档的说明

代码语言：javascript复制
Note: In this guide, we’ll often use the concise Java 8 lambda syntax to specify Java functions, but in older versions of Java you can implement the interfaces in the org.apache.spark.api.java.function package. We describe passing functions to Spark in more detail below.


Spark’s API relies heavily on passing functions in the driver program to run on the cluster. In Java, functions are represented by classes implementing the interfaces in the org.apache.spark.api.java.function package. There are two ways to create such functions:

Implement the Function interfaces in your own class, either as an anonymous inner class or a named one, and pass an instance of it to Spark.
In Java 8, use lambda expressions to concisely define an implementation.

所以如果要完成上面第一种创建方式，在jdk1.8中可以简单的这么写

代码语言：javascript复制
JavaRDD<String> lines = sc.textFile("hdfs://master:9000/testFile/README.md");
JavaRDD<Integer> lineLengths = lines.map(s -> s.length());
int totalLength = lineLengths.reduce((a, b) -> a   b);

要完成第二种方式的创建，简单的这么写

代码语言：javascript复制
List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
JavaRDD<Integer> distData = sc.parallelize(data);

主要不同就是在jdk1.7中我们要自己写一个函数传到map或者reduce方法中，而在jdk1.8中可以直接在map或者reduce方法中写lambda表达式

参考原文：https://blog.csdn.net/tanggao1314/article/details/51570452/

扩展：

SparkContext的parallelize的参数

通过调用SparkContext的parallelize方法，在一个已经存在的Scala集合上创建的（一个Seq对象）。集合的对象将会被拷贝，创建出一个可以被并行操作的分布式数据集。

代码语言：javascript复制
var data = [1, 2, 3, 4, 5]  
var distData = sc.parallelize(data)

在一个Spark程序的开始部分，有好多是用sparkContext的parallelize制作RDD的，是ParallelCollectionRDD，创建一个并行集合。

例如sc.parallelize(0 until numMappers, numMappers)

创建并行集合的一个重要参数，是slices的数目（例子中是numMappers），它指定了将数据集切分为几份。

在集群模式中，Spark将会在一份slice上起一个Task。典型的，你可以在集群中的每个cpu上，起2-4个Slice （也就是每个cpu分配2-4个Task）。

一般来说，Spark会尝试根据集群的状况，来自动设定slices的数目。当让，也可以手动的设置它，通过parallelize方法的第二个参数。（例如：sc.parallelize(data, 10)).

参考：https://blog.csdn.net/caoli98033/article/details/41777065

发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/132515.html原文链接：https://javaforall.cn

spark 大数据 hadoop 分布式 https

0 人点赞