pyspark系列3-spark核心之RDD介绍

浅浅的花香味﹌ 2023-01-17 07:35 182阅读 0赞

### 文章目录 ###

*  一.RDD概念
 *   *  1.1 RDD的特点
     *  1.2 RDD的核心属性
 *  二.操作RDD
 *   *  2.1 PySpark介绍
     *  2.2 PySpark环境配置
     *  2.3 PySpark使用
     *   *  2.3.1 初始化Spark
         *  2.3.2 初始化RDD
         *  2.3.3 RDD操作
         *   *  2.3.3.1 RDD的map操作
             *  2.3.3.1 RDD使用函数
 *  参考:

# 一.RDD概念 #

RDD（resilient distributed dataset ，弹性分布式数据集），是 Spark 中最基础的抽象。它表示了一个可以并行操作的、不可变得、被分区了的元素集合。用户不需要关心底层复杂的抽象处理，直接使用方便的算子处理和计算就可以了。

## 1.1 RDD的特点 ##

**1) . 分布式**  
RDD是一个抽象的概念，RDD在spark driver中，通过RDD来引用数据，数据真正存储在节点机的partition上。

**2). 只读**  
在Spark中RDD一旦生成了，就不能修改。  
那么为什么要设置为只读，设置为只读的话，因为不存在修改，并发的吞吐量就上来了。

**3). 血缘关系**  
我们需要对RDD进行一系列的操作，因为RDD是只读的，我们只能不断的生产新的RDD，这样，新的RDD与原来的RDD就会存在一些血缘关系。

Spark会记录这些血缘关系，在后期的容错上会有很大的益处。

**4). 缓存**  
当一个 RDD 需要被重复使用时，或者当任务失败重新计算的时候，这时如果将 RDD 缓存起来，就可以避免重新计算，保证程序运行的性能。

RDD 的缓存有三种方式：cache、persist、checkPoint。

1.  cache  
    cache 方法不是在被调用的时候立即进行缓存，而是当触发了 action 类型的算子之后，才会进行缓存。
2.  cache 和 persist 的区别  
    其实 cache 底层实际调用的就是 persist 方法，只是缓存的级别默认是 MEMORY\_ONLY，而 persist 方法可以指定其他的缓存级别。
3.  cache 和 checkPoint 的区别  
    checkPoint 是将数据缓存到本地或者 HDFS 文件存储系统中，当某个节点的 executor 宕机了之后，缓存的数据不会丢失，而通过 cache 缓存的数据就会丢掉。

checkPoint 的时候会把 job 从开始重新再计算一遍，因此在 checkPoint 之前最好先进行一步 cache 操作，cache 不需要重新计算，这样可以节省计算的时间。

1.  persist 和 checkPoint 的区别  
    persist 也可以选择将数据缓存到磁盘当中，但是它交给 blockManager 管理的，一旦程序运行结束，blockManager 也会被停止，这时候缓存的数据就会被释放掉。而 checkPoint 持久化的数据并不会被释放，是一直存在的，可以被其它的程序所使用。

## 1.2 RDD的核心属性 ##

RDD 调度和计算都依赖于这五属性  
1). 分区列表  
Spark RDD 是被分区的，每一个分区都会被一个计算任务 (Task) 处理，分区数决定了并行计算的数量，RDD 的并行度默认从父 RDD 传给子 RDD。默认情况下，一个 HDFS 上的数据分片就是一个 partiton，RDD 分片数决定了并行计算的力度，可以在创建 RDD 时指定 RDD 分片个数，如果不指定分区数量，当 RDD 从集合创建时，则默认分区数量为该程序所分配到的资源的 CPU 核数 (每个 Core 可以承载 2~4 个 partition)，如果是从 HDFS 文件创建，默认为文件的 Block 数。

2). 依赖列表  
由于 RDD 每次转换都会生成新的 RDD，所以 RDD 会形成类似流水线一样的前后依赖关系，当然宽依赖就不类似于流水线了，宽依赖后面的 RDD 具体的数据分片会依赖前面所有的 RDD 的所有数据分片，这个时候数据分片就不进行内存中的 Pipeline，一般都是跨机器的，因为有前后的依赖关系，所以当有分区的数据丢失时， Spark 会通过依赖关系进行重新计算，从而计算出丢失的数据，而不是对 RDD 所有的分区进行重新计算。RDD 之间的依赖有两种：窄依赖 ( Narrow Dependency) 和宽依赖 ( Wide Dependency)。RDD 是 Spark 的核心数据结构，通过 RDD 的依赖关系形成调度关系。通过对 RDD 的操作形成整个 Spark 程序。

3). Compute函数，用于计算RDD各分区的值  
每个分区都会有计算函数， Spark 的 RDD 的计算函数是以分片为基本单位的，每个 RDD 都会实现 compute 函数，对具体的分片进行计算，RDD 中的分片是并行的，所以是分布式并行计算，有一点非常重要，就是由于 RDD 有前后依赖关系，遇到宽依赖关系，如 reduce By Key 等这些操作时划分成 Stage， Stage 内部的操作都是通过 Pipeline 进行的，在具体处理数据时它会通过 Blockmanager 来获取相关的数据，因为具体的 split 要从外界读数据，也要把具体的计算结果写入外界，所以用了一个管理器，具体的 split 都会映射成 BlockManager 的 Block，而体的 splt 会被函数处理，函数处理的具体形式是以任务的形式进行的。

4). 分区策略（可选）  
每个 key-value 形式的 RDD 都有 Partitioner 属性，它决定了 RDD 如何分区。当然，Partiton 的个数还决定了每个 Stage 的 Task 个数。RDD 的分片函数可以分区 ( Partitioner)，可传入相关的参数，如 Hash Partitioner 和 Range Partitioner，它本身针对 key- value 的形式，如果不是 key-ale 的形式它就不会有具体的 Partitioner， Partitioner 本身决定了下一步会产生多少并行的分片，同时它本身也决定了当前并行 ( Parallelize) Shuffle 输出的并行数据，从而使 Spak 具有能够控制数据在不同结点上分区的特性，用户可以自定义分区策略，如 Hash 分区等。 spark 提供了 partition By 运算符，能通过集群对 RDD 进行数据再分配来创建一个新的 RDD。

5). 优先位置列表（可选，HDFS实现数据本地化，避免数据移动）  
优先位置列表会存储每个 Partition 的优先位置，对于一个 HDFS 文件来说，就是每个 Partition 块的位置。观察运行 Spark 集群的控制台就会发现， Spark 在具体计算、具体分片以前，它已经清楚地知道任务发生在哪个结点上，也就是说任务本身是计算层面的、代码层面的，代码发生运算之前它就已经知道它要运算的数据在什么地方，有具体结点的信息。这就符合大数据中数据不动代码动的原则。数据不动代码动的最高境界是数据就在当前结点的内存中。这时候有可能是 Memory 级别或 Tachyon 级别的， Spark 本身在进行任务调度时会尽可能地将任务分配到处理数据的数据块所在的具体位置。据 Spark 的 RDD。 Scala 源代码函数 getParferredlocations 可知，每次计算都符合完美的数据本地性。

# 二.操作RDD #

## 2.1 PySpark介绍 ##

PySpark实现了Spark对于Python的API，通过它，用户可以编写运行在Spark之上的Python程序，从而利用到Spark分布式计算的特点。

Python API的实现依赖于Java的API，Python程序端的SparkContext通过py4j调用JavaSparkContext，后者是对Scala的SparkContext的一个封装。而对RDD进行转换和操作的函数由用户通过Python程序来定义，这些函数会被序列化然后发送到各个worker，然后每一个worker启动一个Python进程来执行反序列化之后的函数，通过管道拿到执行之后的结果。

1). Python程序的启动  
和Scala程序一样，Python程序也是通过SparkSubmit提交得以执行，在SparkSubmit中会判断提交的程序是否为Python，如果是，则设置mainClass为PythonRunner。在PythonRunner中，会根据配置选项，以及用户通过命令行提供的–py-files选项，设置好PYTHONPATH，然后启动一个Java的GatewayServer用来被Python程序调用，然后以用户配置的PYSPARK\_PYTHON选项作为Python解释器，执行Python文件，至此用户的Python程序得以启动。

2). SparkContext  
和在Scala中一样，SparkContext是调用Spark进行计算的入口。在Python的context.py中定义了类SparkContext，它封装了一个JavaSparkContext作为它的\_jsc属性。在初始化SparkContext时，首先会调用java\_gateway.py中定义的launch\_gateway方法来初始化JavaGateWay，在launch\_gateway中会引入在Spark中定义的类到SparkContext的属性\_jvm，比如：java\_import(gateway.jvm, “org.apache.spark.SparkConf”)。这样在Python中就可以通过SparkContext.\_jvm.SparkConf引用在Scala中定义的SparkConf这个类，可以实例化这个类的对象，可以调用对象的方法等。在初始化完毕之后，用户就可以调用SparkContext中的方法了，比如textFile和parallelize。

3). RDD  
Python中的RDD对Spark中的RDD进行了一次封装，每一个RDD都对应了一个反序列化的函数。这是因为，尽管在Spark中RDD的元素可以具有任意类型，提供给JavaSparkContext中生成的RDD的只具有Array\[Byte\]类型，也就是说JavaSparkContext的函数返回值是JavaRDD\[Array\[Byte\]\]，这样，Python程序需要把对象先序列化成byte数组，然后把它分布到各个节点进行计算。计算完之后再反序列化成Python的对象。（这其中有一个特殊情况，就是JavaSparkContext返回的是JavaRDD\[String\]，可以把它当成是不需要序列化和反序列化的对象。）在Spark中不需要知道Array\[Byte\]反序列化之后是什么。如何序列化和反序列化、如何对这些Array\[Byte\]进行转换和操作都由Python程序来控制，Spark只是负责资源的调度，负责如何把这些计算分配到各个节点上去执行。

## 2.2 PySpark环境配置 ##

安装好spark后，直接输入pyspark，可调出pyspark工作界面

[root@hp2 ~]# pyspark
    Python 2.7.5 (default, Apr  2 2020, 13:16:51) 
    [GCC 4.8.5 20150623 (Red Hat 4.8.5-39)] on linux2
    Type "help", "copyright", "credits" or "license" for more information.
    Setting default log level to "WARN".
    To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
    21/04/13 08:37:30 WARN lineage.LineageWriter: Lineage directory /var/log/spark/lineage doesn't exist or is not writable. Lineage for this application will be disabled.
    21/04/13 08:37:30 WARN lineage.LineageWriter: Lineage directory /var/log/spark/lineage doesn't exist or is not writable. Lineage for this application will be disabled.
    Welcome to
          ____              __
         / __/__  ___ _____/ /__
        _\ \/ _ \/ _ `/ __/  '_/
       /__ / .__/\_,_/_/ /_/\_\   version 2.4.0-cdh6.3.1
          /_/
    
    Using Python version 2.7.5 (default, Apr  2 2020 13:16:51)
    SparkSession available as 'spark'.
    >>>

1). 引入Python中pyspark工作模块

import pyspark
    from pyspark import SparkContext as sc
    from pyspark import SparkConf
    conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
    sc=SparkContext.getOrCreate(conf)
    #任何Spark程序都是SparkContext开始的，SparkContext的初始化需要一个SparkConf对象，SparkConf包含了Spark集群配置的各种参数(比如主节点的URL)。初始化后，就可以使用SparkContext对象所包含的各种方法来创建和操作RDD和共享变量。Spark shell会自动初始化一个SparkContext(在Scala和Python下可以，但不支持Java)。
    #getOrCreate表明可以视情况新建session或利用已有的session

2). Python脚本执行  
python脚本中需要在开头导入spark相关模块，调用时使用spark-submit提交，如下所示：

spark-submit --master local xxxx.py
    spark-submit --master yarn --deploy-mode cluster xxxx.py
    spark-submit --master yarn --deploy-mode client xxxx.py

## 2.3 PySpark使用 ##

### 2.3.1 初始化Spark ###

编写Spark程序的第一件事情就是创建SparkContext对象，SparkContext负责连接到集群。创建SparkContext先要创建SparkConf对象，该对象可以定义我们Spark程序的相关参数。

conf = SparkConf().setAppName(appName).setMaster(master)
    sc = SparkContext(conf=conf)

其中appName是程序名称，它会显示在集群状态界面上；master是要提交到的集群的地址

### 2.3.2 初始化RDD ###

RDD（Resilient Distributed Datasets）是Spark中抽象出来的弹性分布式数据集，其本质上是一个只读的分区记录集合。每个RDD可以分成多个分区，每个分区就是一个数据集片段。

创建RDD有两种方式：一种是将驱动程序中的已有集合平行化；另外一种是引用外部存储系统的数据集，例如共享文件系统，HDFS, HBase, 或者其他类似Hadoop的数据源.

1). 通过集合创建RDD  
代码:

import pyspark
    from pyspark import SparkContext as sc
    from pyspark import SparkConf
    conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
    sc=SparkContext.getOrCreate(conf)
    
    /* 通过集合创建RDD */
    rdd=sc.parallelize([1,2,3,4,5]) 
    rdd
    -- 查看list被分成了几部分
    rdd.getNumPartitions()
    -- 查看分区状况
    rdd.glom().collect()
    rdd.collect()

测试记录:

[root@hp2 pyspark]# pyspark
    Python 2.7.5 (default, Apr  2 2020, 13:16:51) 
    [GCC 4.8.5 20150623 (Red Hat 4.8.5-39)] on linux2
    Type "help", "copyright", "credits" or "license" for more information.
    Setting default log level to "WARN".
    To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
    21/04/13 10:06:01 WARN cluster.YarnSchedulerBackend$YarnSchedulerEndpoint: Attempted to request executors before the AM has registered!
    Welcome to
          ____              __
         / __/__  ___ _____/ /__
        _\ \/ _ \/ _ `/ __/  '_/
       /__ / .__/\_,_/_/ /_/\_\   version 2.4.0-cdh6.3.1
          /_/
    
    Using Python version 2.7.5 (default, Apr  2 2020 13:16:51)
    SparkSession available as 'spark'.
    >>> 
    >>> import pyspark
    >>> from pyspark import SparkContext as sc
    >>> from pyspark import SparkConf
    >>> conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
    >>> sc=SparkContext.getOrCreate(conf)
    >>> 
    >>> 
    /* 通过集合创建RDD */
    >>> rdd=sc.parallelize([1,2,3,4,5]) 
    >>> rdd
    ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:195
    >>> 
    -- 查看list被分成了几部分
    >>> rdd.getNumPartitions()
    2
    >>> 
    -- 查看分区状况
    >>> rdd.glom().collect()
    [[1, 2], [3, 4, 5]]                                                             
    >>> 
    >>> rdd.collect()
    [1, 2, 3, 4, 5]
    >>>

2). 通过文件创建rdd  
读取一个idcard.txt，获取年龄和性别。

代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    
    from datetime import date
    from pyspark import SparkContext, SparkConf
    import os
    
    #设置环境变量
    os.environ['JAVA_HOME'] = '/usr/java/jdk1.8.0_181'                      # java环境配置
    os.environ['HADOOP_HOME'] = '/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/hadoop'          # hadoop安装目录
    os.environ['SPARK_HOME'] = '/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/spark'  # 设置spark安装目录
    
    today = date.today()
    
    # 设置Spark程序运行的地方，此处设置运行在本地模式，启动2个线程分析数据
    spark_conf = SparkConf().setMaster("local[*]").setAppName("Get Idcard")
    sc = SparkContext(conf = spark_conf)
    
    filename = 'file:///home/pyspark/idcard.txt'
    rdd = sc.textFile(filename)
    
    
    idcards = rdd.collect()
    
    for idcard in idcards:
       # 求出年龄
        my_idcard = idcard
        birh_year = my_idcard[6:10]
    
        age = int(today.year) - int(birh_year)
        print "年龄是:" +  str(age)
    
        # 求性别
        sex_type = my_idcard[-1]
        sex_result = int(sex_type)%2
        if sex_result == 1:
           sex = '男'
        else:
           sex = '女'
    
        print "性别是:" + sex
    
        print("\n")
    
    sc.stop()

测试记录:

[root@hp2 pyspark]# spark-submit test2.py 
    21/04/13 10:45:00 INFO spark.SparkContext: Running Spark version 2.4.0-cdh6.3.1
    21/04/13 10:45:00 INFO logging.DriverLogger: Added a local log appender at: /tmp/spark-5904579a-d4c0-4a0a-8ffe-e7af09931299/__driver_logs__/driver.log
    21/04/13 10:45:00 INFO spark.SparkContext: Submitted application: Get Idcard
    21/04/13 10:45:00 INFO spark.SecurityManager: Changing view acls to: root
    21/04/13 10:45:00 INFO spark.SecurityManager: Changing modify acls to: root
    21/04/13 10:45:00 INFO spark.SecurityManager: Changing view acls groups to: 
    21/04/13 10:45:00 INFO spark.SecurityManager: Changing modify acls groups to: 
    21/04/13 10:45:00 INFO spark.SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users  with view permissions: Set(root); groups with view permissions: Set(); users  with modify permissions: Set(root); groups with modify permissions: Set()
    21/04/13 10:45:00 INFO util.Utils: Successfully started service 'sparkDriver' on port 37239.
    21/04/13 10:45:00 INFO spark.SparkEnv: Registering MapOutputTracker
    21/04/13 10:45:00 INFO spark.SparkEnv: Registering BlockManagerMaster
    21/04/13 10:45:00 INFO storage.BlockManagerMasterEndpoint: Using org.apache.spark.storage.DefaultTopologyMapper for getting topology information
    21/04/13 10:45:00 INFO storage.BlockManagerMasterEndpoint: BlockManagerMasterEndpoint up
    21/04/13 10:45:00 INFO storage.DiskBlockManager: Created local directory at /tmp/blockmgr-6a0ee83c-f223-422a-99f2-66a027cefd55
    21/04/13 10:45:00 INFO memory.MemoryStore: MemoryStore started with capacity 366.3 MB
    21/04/13 10:45:00 INFO spark.SparkEnv: Registering OutputCommitCoordinator
    21/04/13 10:45:01 INFO util.log: Logging initialized @1715ms
    21/04/13 10:45:01 INFO server.Server: jetty-9.3.z-SNAPSHOT, build timestamp: 2018-09-05T05:11:46+08:00, git hash: 3ce520221d0240229c862b122d2b06c12a625732
    21/04/13 10:45:01 INFO server.Server: Started @1786ms
    21/04/13 10:45:01 INFO server.AbstractConnector: Started ServerConnector@2a2e00fb{HTTP/1.1,[http/1.1]}{0.0.0.0:4040}
    21/04/13 10:45:01 INFO util.Utils: Successfully started service 'SparkUI' on port 4040.
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@52fd23a9{/jobs,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@524f7a39{/jobs/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@1e124efa{/jobs/job,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@7eee02a9{/jobs/job/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@57d931ff{/stages,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@5b30c060{/stages/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@1dad9173{/stages/stage,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@507e4c6a{/stages/stage/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@21a9791f{/stages/pool,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@733a2714{/stages/pool/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@7b51b4f5{/storage,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@596eaae9{/storage/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@3365515a{/storage/rdd,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@1145c2a5{/storage/rdd/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@67c53a64{/environment,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@4b50fd97{/environment/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@52269a26{/executors,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@6dc93a2d{/executors/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@23113bea{/executors/threadDump,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@6f61d65a{/executors/threadDump/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@970863a{/static,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@4a6109c2{/,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@62b52167{/api,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@583d3e41{/jobs/job/kill,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@48f167f0{/stages/stage/kill,null,AVAILABLE,@Spark}
    21/04/13 10:45:01 INFO ui.SparkUI: Bound SparkUI to 0.0.0.0, and started at http://hp2:4040
    21/04/13 10:45:01 INFO executor.Executor: Starting executor ID driver on host localhost
    21/04/13 10:45:01 INFO util.Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 44348.
    21/04/13 10:45:01 INFO netty.NettyBlockTransferService: Server created on hp2:44348
    21/04/13 10:45:01 INFO storage.BlockManager: Using org.apache.spark.storage.RandomBlockReplicationPolicy for block replication policy
    21/04/13 10:45:01 INFO storage.BlockManagerMaster: Registering BlockManager BlockManagerId(driver, hp2, 44348, None)
    21/04/13 10:45:01 INFO storage.BlockManagerMasterEndpoint: Registering block manager hp2:44348 with 366.3 MB RAM, BlockManagerId(driver, hp2, 44348, None)
    21/04/13 10:45:01 INFO storage.BlockManagerMaster: Registered BlockManager BlockManagerId(driver, hp2, 44348, None)
    21/04/13 10:45:01 INFO storage.BlockManager: external shuffle service port = 7337
    21/04/13 10:45:01 INFO storage.BlockManager: Initialized BlockManager: BlockManagerId(driver, hp2, 44348, None)
    21/04/13 10:45:01 INFO handler.ContextHandler: Started o.s.j.s.ServletContextHandler@29f4468f{/metrics/json,null,AVAILABLE,@Spark}
    21/04/13 10:45:02 INFO scheduler.EventLoggingListener: Logging events to hdfs://nameservice1/user/spark/applicationHistory/local-1618281901231
    21/04/13 10:45:02 INFO spark.SparkContext: Registered listener com.cloudera.spark.lineage.NavigatorAppListener
    21/04/13 10:45:02 INFO logging.DriverLogger$DfsAsyncWriter: Started driver log file sync to: /user/spark/driverLogs/local-1618281901231_driver.log
    21/04/13 10:45:02 INFO memory.MemoryStore: Block broadcast_0 stored as values in memory (estimated size 290.4 KB, free 366.0 MB)
    21/04/13 10:45:02 INFO memory.MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 28.4 KB, free 366.0 MB)
    21/04/13 10:45:02 INFO storage.BlockManagerInfo: Added broadcast_0_piece0 in memory on hp2:44348 (size: 28.4 KB, free: 366.3 MB)
    21/04/13 10:45:02 INFO spark.SparkContext: Created broadcast 0 from textFile at NativeMethodAccessorImpl.java:0
    21/04/13 10:45:02 INFO mapred.FileInputFormat: Total input files to process : 1
    21/04/13 10:45:02 INFO spark.SparkContext: Starting job: collect at /home/pyspark/test2.py:23
    21/04/13 10:45:02 INFO scheduler.DAGScheduler: Got job 0 (collect at /home/pyspark/test2.py:23) with 2 output partitions
    21/04/13 10:45:02 INFO scheduler.DAGScheduler: Final stage: ResultStage 0 (collect at /home/pyspark/test2.py:23)
    21/04/13 10:45:02 INFO scheduler.DAGScheduler: Parents of final stage: List()
    21/04/13 10:45:02 INFO scheduler.DAGScheduler: Missing parents: List()
    21/04/13 10:45:02 INFO scheduler.DAGScheduler: Submitting ResultStage 0 (file:///home/pyspark/idcard.txt MapPartitionsRDD[1] at textFile at NativeMethodAccessorImpl.java:0), which has no missing parents
    21/04/13 10:45:02 INFO memory.MemoryStore: Block broadcast_1 stored as values in memory (estimated size 3.4 KB, free 366.0 MB)
    21/04/13 10:45:03 INFO memory.MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 2.0 KB, free 366.0 MB)
    21/04/13 10:45:03 INFO storage.BlockManagerInfo: Added broadcast_1_piece0 in memory on hp2:44348 (size: 2.0 KB, free: 366.3 MB)
    21/04/13 10:45:03 INFO spark.SparkContext: Created broadcast 1 from broadcast at DAGScheduler.scala:1164
    21/04/13 10:45:03 INFO scheduler.DAGScheduler: Submitting 2 missing tasks from ResultStage 0 (file:///home/pyspark/idcard.txt MapPartitionsRDD[1] at textFile at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0, 1))
    21/04/13 10:45:03 INFO scheduler.TaskSchedulerImpl: Adding task set 0.0 with 2 tasks
    21/04/13 10:45:03 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, executor driver, partition 0, PROCESS_LOCAL, 7889 bytes)
    21/04/13 10:45:03 INFO scheduler.TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1, localhost, executor driver, partition 1, PROCESS_LOCAL, 7889 bytes)
    21/04/13 10:45:03 INFO executor.Executor: Running task 0.0 in stage 0.0 (TID 0)
    21/04/13 10:45:03 INFO executor.Executor: Running task 1.0 in stage 0.0 (TID 1)
    21/04/13 10:45:03 INFO rdd.HadoopRDD: Input split: file:/home/pyspark/idcard.txt:0+104
    21/04/13 10:45:03 INFO rdd.HadoopRDD: Input split: file:/home/pyspark/idcard.txt:104+105
    21/04/13 10:45:03 INFO executor.Executor: Finished task 0.0 in stage 0.0 (TID 0). 867 bytes result sent to driver
    21/04/13 10:45:03 INFO executor.Executor: Finished task 1.0 in stage 0.0 (TID 1). 848 bytes result sent to driver
    21/04/13 10:45:03 INFO scheduler.TaskSetManager: Finished task 1.0 in stage 0.0 (TID 1) in 90 ms on localhost (executor driver) (1/2)
    21/04/13 10:45:03 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 107 ms on localhost (executor driver) (2/2)
    21/04/13 10:45:03 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
    21/04/13 10:45:03 INFO scheduler.DAGScheduler: ResultStage 0 (collect at /home/pyspark/test2.py:23) finished in 0.168 s
    21/04/13 10:45:03 INFO scheduler.DAGScheduler: Job 0 finished: collect at /home/pyspark/test2.py:23, took 0.232997 s
    年龄是:52
    性别是:女
    
    
    年龄是:43
    性别是:男
    
    
    年龄是:52
    性别是:男
    
    
    年龄是:45
    性别是:女
    
    
    年龄是:58
    性别是:女
    
    
    年龄是:42
    性别是:女
    
    
    年龄是:65
    性别是:女
    
    
    年龄是:52
    性别是:女
    
    
    年龄是:41
    性别是:女
    
    
    年龄是:52
    性别是:男
    
    
    年龄是:31
    性别是:男
    
    
    21/04/13 10:45:03 INFO server.AbstractConnector: Stopped Spark@2a2e00fb{HTTP/1.1,[http/1.1]}{0.0.0.0:4040}
    21/04/13 10:45:03 INFO ui.SparkUI: Stopped Spark web UI at http://hp2:4040
    21/04/13 10:45:03 INFO spark.MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
    21/04/13 10:45:03 INFO memory.MemoryStore: MemoryStore cleared
    21/04/13 10:45:03 INFO storage.BlockManager: BlockManager stopped
    21/04/13 10:45:03 INFO storage.BlockManagerMaster: BlockManagerMaster stopped
    21/04/13 10:45:03 INFO scheduler.OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
    21/04/13 10:45:03 INFO spark.SparkContext: Successfully stopped SparkContext
    21/04/13 10:45:03 INFO util.ShutdownHookManager: Shutdown hook called
    21/04/13 10:45:03 INFO util.ShutdownHookManager: Deleting directory /tmp/spark-dd18587e-5244-4b4e-9eb1-f147c18f1306
    21/04/13 10:45:03 INFO util.ShutdownHookManager: Deleting directory /tmp/spark-5904579a-d4c0-4a0a-8ffe-e7af09931299/pyspark-16b22e0d-ced4-4f30-955e-c0601e06c1ac
    21/04/13 10:45:03 INFO util.ShutdownHookManager: Deleting directory /tmp/spark-5904579a-d4c0-4a0a-8ffe-e7af09931299
    [root@hp2 pyspark]#

### 2.3.3 RDD操作 ###

RDDs支持两种类型的操作：一种是转换(transformations), 该操作从已有数据集创建新的数据集；另外一种是动作(actions)，该操作在数据集上执行计算之后返回一个值给驱动程序。例如, map就是一个转换，这个操作在数据集的每个元素上执行一个函数并返回一个处理之后新的RDD结果。另一方面，reduce是一个动作，这个操作按照某个函数规则聚集RDD中的所有元素并且把最终结果返回给驱动程序。

Spark中的所有转换操作都是lazy模式的，也就是说，不是立马做转换计算结果，而是将这些转换操作记录在相应的数据集上，当需要通过动作(action)把结果返回给驱动程序时才真正执行。这个设计使Spark运行起来更加高效。例如，如果通过map创建的数据集后续会被reduce用到，那么只有reduce的结果会返回给驱动程序，而不是更大的map结果。默认情况下，RDD上的转换操作在每次做动作时，都会重新执行计算一次。然而，我们可以使用persist(或者cache)函数将RDD存放在内存中，方便后续的快速访问。另外，Spark也支持将RDD存放在磁盘上，或者在多个节点冗余存储。

**常见Transformations操作**

<table> 
 <thead> 
  <tr> 
   <th>Transformation</th> 
   <th>含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>map(func)</td> 
   <td>对每个RDD元素应用func之后，构造成新的RDD</td> 
  </tr> 
  <tr> 
   <td>filter(func)</td> 
   <td>对每个RDD元素应用func, 将func为true的元素构造成新的RDD</td> 
  </tr> 
  <tr> 
   <td>flatMap(func)</td> 
   <td>和map类似,但是flatMap可以将一个输出元素映射成0个或多个元素。(也就是说func返回的是元素序列而不是单个元素).</td> 
  </tr> 
  <tr> 
   <td>mapPartitions(func)</td> 
   <td>和map类似，但是在RDD的不同分区上独立执行。所以函数func的参数是一个Python迭代器，输出结果也应该是迭代器【即func作用为Iterator =&gt; Iterator】</td> 
  </tr> 
  <tr> 
   <td>mapPartitionsWithIndex(func)</td> 
   <td>和mapPartitions类似, but also provides func with an integer value representing the index of the partition, 但是还为函数func提供了一个正式参数，用来表示分区的编号。【此时func作用为(Int, Iterator) =&gt; Iterator 】</td> 
  </tr> 
  <tr> 
   <td>sample(withReplacement, fraction, seed)</td> 
   <td>抽样: fraction是抽样的比例0~1之间的浮点数; withRepacement表示是否有放回抽样, True是有放回, False是无放回；seed是随机种子。</td> 
  </tr> 
  <tr> 
   <td>union(otherDataset)</td> 
   <td>并集操作，重复元素会保留（可以通过distinct操作去重）</td> 
  </tr> 
  <tr> 
   <td>intersection(otherDataset)</td> 
   <td>交集操作，结果不会包含重复元素</td> 
  </tr> 
  <tr> 
   <td>distinct([numTasks]))</td> 
   <td>去重操作</td> 
  </tr> 
  <tr> 
   <td>groupByKey([numTasks])</td> 
   <td>把Key相同的数据放到一起【(K, V) =&gt; (K, Iterable)】，需要注意的问题：1. 如果分组(grouping)操作是为了后续的聚集(aggregation)操作(例如sum/average)， 使用reduceByKey或者aggregateByKey更高效。2.默认情况下，并发度取决于分区数量。我们可以传入参数numTasks来调整并发任务数。</td> 
  </tr> 
  <tr> 
   <td>reduceByKey(func, [numTasks])</td> 
   <td>首先按Key分组，然后将相同Key对应的所有Value都执行func操作得到一个值。func必须是(V, V) =&gt; V’的计算操作。numTasks作用跟上面提到的groupByKey一样。</td> 
  </tr> 
  <tr> 
   <td>sortByKey([ascending], [numTasks])</td> 
   <td>按Key排序。通过第一个参数True/False指定是升序还是降序。</td> 
  </tr> 
  <tr> 
   <td>join(otherDataset, [numTasks])</td> 
   <td>类似SQL中的连接(内连接)，即(K, V) and (K, W) =&gt; (K, (V, W))，返回所有连接对。外连接通过:leftOUterJoin(左出现右无匹配为空)、rightOuterJoin（右全出现左无匹配为空）、fullOuterJoin实现（左右全出现无匹配为空）。</td> 
  </tr> 
  <tr> 
   <td>cogroup(otherDataset, [numTasks])</td> 
   <td>对两个RDD做groupBy。即(K, V) and (K, W) =&gt; (K, Iterable, Iterable(W))。别名groupWith。</td> 
  </tr> 
  <tr> 
   <td>pipe(command, [envVars])</td> 
   <td>将驱动程序中的RDD交给shell处理（外部进程），例如Perl或bash脚本。RDD元素作为标准输入传给脚本，脚本处理之后的标准输出会作为新的RDD返回给驱动程序。</td> 
  </tr> 
  <tr> 
   <td>coalesce(numPartitions)</td> 
   <td>将RDD的分区数减小到numPartitions。当数据集通过过滤减小规模时，使用这个操作可以提升性能。</td> 
  </tr> 
  <tr> 
   <td>repartition(numPartitions)</td> 
   <td>将数据重新随机分区为numPartitions个。这会导致整个RDD的数据在集群网络中洗牌。</td> 
  </tr> 
  <tr> 
   <td>repartitionAndSortWithinPartitions(partitioner)</td> 
   <td>使用partitioner函数充分去，并在分区内排序。这比先repartition然后在分区内sort高效，原因是这样迫使排序操作被移到了shuffle阶段。</td> 
  </tr> 
 </tbody> 
</table>

**常见Actions操作**

<table> 
 <thead> 
  <tr> 
   <th>Action</th> 
   <th>含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>reduce(func)</td> 
   <td>使用func函数聚集RDD中的元素（func接收两个参数返回一个值）。这个函数应该满足结合律和交换律以便能够正确并行计算。</td> 
  </tr> 
  <tr> 
   <td>collect()</td> 
   <td>将RDD转为数组返回给驱动程序。这个在执行filter等操作之后返回足够小的数据集是比较有用。</td> 
  </tr> 
  <tr> 
   <td>count()</td> 
   <td>返回RDD中的元素数量。</td> 
  </tr> 
  <tr> 
   <td>first()</td> 
   <td>返回RDD中的第一个元素。（同take(1))</td> 
  </tr> 
  <tr> 
   <td>take(n)</td> 
   <td>返回由RDD的前N个元素组成的数组。</td> 
  </tr> 
  <tr> 
   <td>takeSample(withReplacement, num, [seed])</td> 
   <td>返回num个元素的数组，这些元素抽样自RDD，withReplacement表示是否有放回，seed是随机数生成器的种子）。</td> 
  </tr> 
  <tr> 
   <td>takeOrdered(n, [ordering])</td> 
   <td>返回RDD的前Ｎ个元素，使用自然顺序或者通过ordering函数对将个元素转换为新的Key.</td> 
  </tr> 
  <tr> 
   <td>saveAsTextFile(path)</td> 
   <td>将RDD元素写入文本文件。Spark自动调用元素的toString方法做字符串转换。</td> 
  </tr> 
  <tr> 
   <td>saveAsSequenceFile(path)(Java and Scala)</td> 
   <td>将RDD保存为Hadoop SequenceFile.这个过程机制如下：1. Pyrolite用来将序列化的Python RDD转为Java对象RDD;2. Java RDD中的Key/Value被转为Writable然后写到文件。</td> 
  </tr> 
  <tr> 
   <td>countByKey()</td> 
   <td>统计每个Key出现的次数，只对(K, V)类型的RDD有效，返回(K, int)词典。</td> 
  </tr> 
  <tr> 
   <td>foreach(func)</td> 
   <td>在所有RDD元素上执行函数func。</td> 
  </tr> 
 </tbody> 
</table>

![image.png][]

上图是关于Transformations和Actions的一个介绍图解，可以发现，Transformations操作过后的RDD依旧是RDD，而Actions过后的RDD都是非RDD。

#### 2.3.3.1 RDD的map操作 ####

求txt文档文本总长度

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark import SparkContext, SparkConf
    
    # 设置Spark程序运行的地方，此处设置运行在本地模式，启动2个线程分析数据
    spark_conf = SparkConf().setMaster("local[*]").setAppName("Get Idcard")
    sc = SparkContext(conf = spark_conf)
    
    filename = 'file:///home/pyspark/idcard.txt'
    rdd = sc.textFile(filename)
    
    # 求每一行的长度
    rdd_length = rdd.map(lambda s: len(s))
    # 求所有行的长度
    totalLength = rdd_length.reduce(lambda a, b: a + b)
    
    print(totalLength)
    sc.stop()

#### 2.3.3.1 RDD使用函数 ####

上一列，只是将len函数封装成一个函数

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from pyspark import SparkContext, SparkConf
    
    # 定义一个函数,求长度
    def myFunc(s):
        return len(s)
    
    # 设置Spark程序运行的地方，此处设置运行在本地模式，启动2个线程分析数据
    spark_conf = SparkConf().setMaster("local[*]").setAppName("Get Idcard")
    sc = SparkContext(conf = spark_conf)
    
    filename = 'file:///home/pyspark/idcard.txt'
    rdd = sc.textFile(filename)
    
    # 求每一行的长度
    rdd_length = rdd.map(myFunc)
    # 求所有行的长度
    totalLength = rdd_length.reduce(lambda a, b: a + b)
    
    print(totalLength)
    sc.stop()

# 参考: #

1.http://spark.apache.org/docs/latest/rdd-programming-guide.html  
2.https://www.modb.pro/db/45929  
3.https://gourderwa.blog.csdn.net/article/details/104350323  
4.http://spark.apache.org/docs/latest/api/python/reference/pyspark.html\#rdd-apis  
5.https://www.jianshu.com/p/321034864bdb/  
6.https://vlambda.com/wz\_7iNyAJSkOIK.html

[image.png]: /images/20221021/4b622944c4c44e2a8da74a2624549bfe.png