Spark RDD简单操作

ゝ一纸荒年。 2022-04-12 02:30 270阅读 0赞

# Spark RDD操作 #

spark快速大数据分析.pdf[下载][Link 1]：[https://download.csdn.net/download/u014646662/10816588][Link 1]

[IDEA 创建scala spark的Mvn项目][IDEA _scala spark_Mvn]：[https://blog.csdn.net/u014646662/article/details/84618032][IDEA _scala spark_Mvn]

1、Spark RDD转化操作

2、Spark RDD行动操作

3、惰性求值

对人工智能感兴趣的同学，可以点击以下链接：

[现在人工智能非常火爆，很多朋友都想学，但是一般的教程都是为博硕生准备的，太难看懂了。最近发现了一个非常适合小白入门的教程，不仅通俗易懂而且还很风趣幽默。所以忍不住分享一下给大家。点这里可以跳转到教程。][Link 2]

[https://www.cbedai.net/u014646662][Link 2]

RDD 支持两种操作：转化操作和行动操作。

RDD 的转化操作是返回一个新的RDD 的操作，比如map() 和filter()，而行动操作则是向驱动器程序返回结果或把结果写入外部系统的操作，会触发实际的计算，比如count() 和first()。Spark 对待转化操作和行动操作的方式很不一样，因此理解你正在进行的操作的类型是很重要的。如果对于一个特定的函数是属于转化操作还是行动操作感到困惑，你可以看看它的返回值类型：转化操作返回的是RDD，而行动操作返回的是其他的数据类型。

## Spark RDD转化操作 ##

RDD 的转化操作是返回新RDD 的操作。转化出来的RDD 是惰性求值的，只有在行动操作中用到这些RDD 时才会被计算。许多转化操作都是针对各个元素的，也就是说，这些转化操作每次只会操作RDD 中的一个元素。不过并不是所有的转化操作都是这样的。举个例子，我们筛选默认使用bash的用户

注意，filter() 操作不会改变已有的users 中的数据。实际上，该操作会返回一个全新的RDD。users在后面的程序中还可以继续使用，比如我们还可以从中搜索别的单词。

接下来，我们使用另一个转化操作union() 来打印出包含bash或nologin的行数。

scala> var users = sc.textFile("passwd")
    users: org.apache.spark.rdd.RDD[String] = passwd MapPartitionsRDD[12] at textFile at <console>:27
    
    scala> var bashUser = users.filter(line => line.contains("/bin/bash"))
    bashUser: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[13] at filter at <console>:29
    
    scala> var nologins = users.filter(line => line.contains("/sbin/nologin"))
    nologins: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[14] at filter at <console>:29
    
    scala> var bash_nologin = bashUser.union(nologins)
    bash_nologin: org.apache.spark.rdd.RDD[String] = UnionRDD[15] at union at <console>:33

union() 与filter() 的不同点在于它操作两个RDD 而不是一个。转化操作可以操作任意数量的输入RDD。

最后要说的是，通过转化操作，你从已有的RDD 中派生出新的RDD，Spark 会使用谱系图（lineage graph）来记录这些不同RDD 之间的依赖关系。Spark 需要用这些信息来按需计算每个RDD，也可以依靠谱系图在持久化的RDD 丢失部分数据时恢复所丢失的数据。（对bash\_nologin 做一个行动操作就会生成下面的图）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2NDY2NjI_size_16_color_FFFFFF_t_70][]

## Spark RDD行动操作 ##

转化操作是不会自动做计算的，只有执行力行动操作才会运算

我们继续对上述例子操作：首先统计个数

scala> bashUser.count
    res19: Long = 10
    
    scala> nologins.count
    res20: Long = 37
    
    scala> bash_nologin.count
    res21: Long = 47

三个统计显示：默认使用bash的用户有10个，不可登录的用户37个，两者合并后47个

使用take收集一些元素

scala> bash_nologin.take(15).foreach(println)
    root:x:0:0:root:/root:/bin/bash
    hdfs:x:492:491:Hadoop HDFS:/var/lib/hadoop-hdfs:/bin/bash
    yarn:x:491:490:Hadoop Yarn:/var/lib/hadoop-yarn:/bin/bash
    impala:x:490:489:Impala:/var/lib/impala:/bin/bash
    mapred:x:489:488:Hadoop MapReduce:/var/lib/hadoop-mapreduce:/bin/bash
    mysql:x:27:27:MySQL Server:/var/lib/mysql:/bin/bash
    kms:x:482:482:Hadoop KMS:/var/lib/hadoop-kms:/bin/bash
    llama:x:500:481:Llama:/var/lib/llama:/bin/bash
    httpfs:x:481:480:Hadoop HTTPFS:/var/lib/hadoop-httpfs:/bin/bash
    cloudera:x:501:501::/home/cloudera:/bin/bash
    bin:x:1:1:bin:/bin:/sbin/nologin
    daemon:x:2:2:daemon:/sbin:/sbin/nologin
    adm:x:3:4:adm:/var/adm:/sbin/nologin
    lp:x:4:7:lp:/var/spool/lpd:/sbin/nologin
    mail:x:8:12:mail:/var/spool/mail:/sbin/nologin

take()只能 获取了RDD 中的少量元素，而collect可以获取RRD所有元素

scala> bash_nologin.collect().foreach(println)
    root:x:0:0:root:/root:/bin/bash
    hdfs:x:492:491:Hadoop HDFS:/var/lib/hadoop-hdfs:/bin/bash
    yarn:x:491:490:Hadoop Yarn:/var/lib/hadoop-yarn:/bin/bash
    ......

注意：collect() 不能用在大规模数据集上。他会把数据放到单台机器的内存中

当RDD特别大时，可以把RDD中的数据存到文件中，例如存储到文本文档saveAsTextFile()

bash_nologin.saveAsTextFile("bash_nologin")

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2NDY2NjI_size_16_color_FFFFFF_t_70 1][]

### 惰性求值 ###

前面提过，RDD 的转化操作都是惰性求值的。这意味着在被调用行动操作之前Spark 不会开始计算。这对新用户来说可能与直觉有些相违背之处，但是对于那些使用过诸如Haskell等函数式语言或者类似LINQ 这样的数据处理框架的人来说，会有些似曾相识。

惰性求值意味着当我们对RDD 调用转化操作（例如调用map()）时，操作不会立即执行。相反，Spark 会在内部记录下所要求执行的操作的相关信息。我们不应该把RDD 看作存放着特定数据的数据集，而最好把每个RDD 当作我们通过转化操作构建出来的、记录如何计算数据的指令列表。把数据读取到RDD 的操作也同样是惰性的。因此，当我们调用sc.textFile() 时，数据并没有读取进来，而是在必要时才会读取。和转化操作一样的是，读取数据的操作也有可能会多次执行。

虽然转化操作是惰性求值的，但还是可以随时通过运行一个行动操作来强制Spark 执行RDD 的转化操作，比如使用count()。这是一种对你所写的程序进行部分测试的简单方法。

Spark 使用惰性求值，这样就可以把一些操作合并到一起来减少计算数据的步骤。在类似Hadoop MapReduce 的系统中，开发者常常花费大量时间考虑如何把操作组合到一起，以减少MapReduce 的周期数。而在Spark 中，写出一个非常复杂的映射并不见得能比使用很多简单的连续操作获得好很多的性能。因此，用户可以用更小的操作来组织他们的程序，这样也使这些操作更容易管理。

[Link 1]: https://download.csdn.net/download/u014646662/10816588
[IDEA _scala spark_Mvn]: https://blog.csdn.net/u014646662/article/details/84618032
[Link 2]: https://www.cbedai.net/u014646662
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2NDY2NjI_size_16_color_FFFFFF_t_70]: /images/20220412/239a80a1fbb745db8bf8a916b384477e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2NDY2NjI_size_16_color_FFFFFF_t_70 1]: /images/20220412/e1cbbd6786e14fb495a3b3d0c0244b57.png