【PySpark学习笔记一】弹性分布式数据集RDD

不念不忘少年蓝@ 2022-05-09 15:22 214阅读 0赞

*  更新时间：2018-10-16

### RDD的内部运行方式 ###

Spark优势：每个转换操作并行执行，大大提高速度。

数据集的转换通常是惰性的，即在transformation过程不会执行程序，只有在action过程才会执行。

### 创建RDD ###

导入相关程序库

from pyspark import SparkContext as sc
    from pyspark import SparkConf

创建RDD

# 将list或array转为RDD
    data = sc.parallelize(
    [("A", 2), ("B", 3), ("C", 4)])
    # 从本地或外部文件导入
    data_from_file = sc.textFile(path, n) # n为分区

#### Schema ####

RDD是无Schema的数据结构，因此可以混合使用tuple、dict、list等多种数据结构。

#### 全局作用域与局部作用域 ####

Spark可以在两种模式下运行：本地模式和集群模式。

### transformation操作 ###

#### .map(…) ####

该方法应用在每一个RDD元素上，与python中的`map`函数类似：

data_new = data.map(lambda row: (row,1))

#### .filter(…) ####

该方法对RDD中元素按照条件进行过滤：

data_filtered = data.filter(lambda row: row[1] == 2 or row[1] == 3)

#### .flatMap(…) ####

`.flatMap(…)`方法和`.map(...)`方法类似，但`.flatMap(…)`方法返回一个扁平结果，而不是一个列表。

data_flat = data.flatMap(lambda row: row[1]+1)

#### .distinct(…) ####

该方法返回指定列中不同值的列表：

data_distinct = data.map(lambda row: (row,1)).distinct()

#### .sample(…) ####

该方法返回数据集的随机样本，第一个参数指定采样是否应该替换，第二个参数定义返回数据的分数，第三个参数是伪随机数产生器的种子。

fraction = 0.1
    data_sample = data.sample(False, fraction, 666)

上例中，从原数据集中随机抽样10%。

#### .leftOuterJoin(…) ####

与SQL中`left join`的语法类似，用以链接两个RDD：

rdd1 = sc.parallelize([('a', 1), ('b', 4), ('c', 10)])
    rdd2 = sc.parallelize([('a', 4), ('a', 1), ('b', '6'), ('d', 11)])
    rdd3 = rdd1.leftOuterJoin(rdd2)

运行结果如下：

rdd3.collect()
    # Out[1]: [('c',(10, None)), ('b', (4,'6')), ('a', (1,4)), ('a', (1,1))]

此外，还有`.join(...)`和`.intersection(...)`方法，均与SQL中意义相同。

#### .repartition(…) ####

该方法重新对数据集进行分区，改变数据集分区的数量：

rdd1 = rdd1.repartition(4)

上例中，将`rdd1`重新设置了4个分区。

### Action操作 ###

#### .take(…) ####

该方法用于返回RDD中单个分区的前n行元素：

rdd1.take(1)

如果想要随机的记录，可以使用`takeSample(...)`方法，这个方法有三个参数：第一个参数表示采样是否应该被替换，第二个参数指定要返回的记录数量，第三个参数是伪随机数发生器种子。

data_take_sample = data.takeSample(False, 1, 667)

#### .collect(…) ####

与`.take(...)`方法类似，但`.collect(...)`方法返回全部RDD元素，因此在大规模RDD时不推荐使用。

#### .reduce(…) ####

`.reduce(...)`方法使用指定的方法减少RDD中的元素。

rdd1.map(lambda row: row[1]).reduce(lambda x,y: x+y)

上述方法用于计算RDD总的元素数量，首先通过`.map(...)`转换，创建一个包含rdd1所有值的列表，然后使用`.reduce(...)`方法对结果进行处理。

`.reduceByKey(...)`方法和`.reduce(...)`方法类似，但`.reduceByKey(...)`是在键-键基础上进行：

data_key = sc.parallelize([('a',4),('b',3),('c',2),('a',3),('d',2),('b',5),('d',1)], 4)
    data_key.reduceByKey(lambda x, y: x + y).collect()
    #Out[2]: [('b',8),('a',7),('c',2),('d',3)]

#### .count(…) ####

该方法用于统计RDD中元素数量，如果数据集是k-v格式的，可以使用`countByKey(...)`获取不同键的计数：

data.count()
    data.countByKey().items()

#### .saveAsTextFile(…) ####

该方法将RDD保存为文件，每个文件一个分区：

data.saveAsTextFile(path + filename)

发表评论取消回复

表情：

评论列表（有 0 条评论，214人围观）

还没有评论，来说两句吧...

相关阅读

相关 Spark RDD弹性分布式数据集

> 目录 > > RDD简介 > > 一、RDD的定义 > > 二、RDD的5大特征 > > RDD的创建方式 > > 一、从文件系统加载数据创建RDD >

素颜马尾好姑娘i/ 2024年04月08日 08:25/ 0 赞/ 67 阅读

相关 Spark弹性式数据集RDDs

-------------------- title: Spark弹性式数据集RDDs date: 2021-05-08 16:24:20 tags: Spa

落日映苍穹つ/ 2023年10月04日 12:49/ 0 赞/ 14 阅读

相关 Spark之RDD(弹性分布式数据集) 详解

1.RDD简述 RDD简单的解释： RDD是将数据项拆分为多个分区的集合,存储在集群的工作节点上的内存和磁盘中，并执行正确的操作。复杂的解释： RDD是用于

￡神魔★判官ぃ/ 2023年03月05日 12:56/ 0 赞/ 6 阅读

相关 Spark RDD弹性分布数据集详解

RDD简介 RDD是Spark提供的最重要的抽象概念，我们可以将RDD理解为一个分布式存储在集群中的大型数据集合，不同RDD之间可以通过转换操作形成依赖关系实现管道化，从而避

- 日理万妓/ 2023年02月12日 09:23/ 0 赞/ 8 阅读

相关 Spark：RDD弹性分布式数据集——实践

RDD弹性分布式数据集——实践一、预先准备二、RDD创建 2.1 从本地文件系统中加载数据创建RDD

我就是我/ 2022年11月17日 13:53/ 0 赞/ 338 阅读

相关 Spark RDD弹性分布式数据集——理论

Spark RDD 理论 1.1 RDD简介 1.2 RDD五大特征 1.3 RDD的处理过程 1.4 RDD的分区

傷城~/ 2022年11月17日 10:15/ 0 赞/ 250 阅读

相关 Spark RDD(分布式弹性数据集)

分布式弹性数据集(RDD)是一个不可变型分布式对象集合，这些元素分布在集群中的多个节点上。RDD是Spark中使用的基本对象。他们是不可变型集合，用于表示数据，并且内置了可靠性

￡神魔★判官ぃ/ 2022年08月31日 09:55/ 0 赞/ 166 阅读

相关 Spark核心类：弹性分布式数据集RDD及其转换和操作pyspark.RDD

http://[blog.csdn.net/pipisorry/article/details/53257188][blog.csdn.net_pipisorry_articl

Myth丶恋晨/ 2022年07月14日 00:17/ 0 赞/ 180 阅读

相关【PySpark学习笔记一】弹性分布式数据集RDD

更新时间：2018-10-16 RDD的内部运行方式 Spark优势：每个转换操作并行执行，大大提高速度。数据集的转换通常是惰性的，即在transformat

不念不忘少年蓝@/ 2022年05月09日 15:22/ 0 赞/ 215 阅读

相关 PySpark 基础知识-RDD 弹性分布式数据集和 DataFrame

1.RDD（弹性分布式数据集）创建第一种：读取一个外部数据集。比如，从本地文件加载数据集，或者从HDFS文件系统、HBase、Amazon S3等外部数据源中加载数据

系统管理员/ 2022年01月21日 23:19/ 0 赞/ 396 阅读