1200字范文 > 学习笔记Spark（四）—— Spark编程基础（创建RDD RDD算子文件读取与存储）

学习笔记Spark（四）—— Spark编程基础（创建RDD RDD算子文件读取与存储）

时间：2024-05-26 17:49:10

文章目录

一、创建RDD1.1、启动Spark shell1.2、创建RDD1.2.1、从集合中创建RDD1.2.2、从外部存储中创建RDD任务1：二、RDD算子2.1、map与flatMap算子应用2.1.1、map2.1.2、flatMap2.1.3、mapPartitions2.2、sortBy与filter算子应用2.2.1、sortBy2.2.2、filter任务2：2.3、交集与并集计算的算子应用2.3.1、distinct2.3.2、union2.3.3、intersection2.3.4、subtract2.3.5、cartesian任务3：2.4、键值对RDD常用算子2.4.1、创建键值对RDD2.4.2、mapValues2.4.3、groupByKey2.4.4、reduceByKey2.4.5、join2.5、常用Action类型算子2.5.1、lookup2.5.2、collect2.5.3、take2.5.4、count任务4:三、文件读取与存储3.1、saveAsTextFile3.2、repartition3.3、saveAsSequenceFile3.4、sequenceFile综合练习1

一、创建RDD

1.1、启动Spark shell

进入Spark命令行交互界面：spark-shell

退出交互界面：:q

查看客户端：（http://master:8080）

设置日志级别

sc.setLogLevel("INFO")sc.setLogLevel("WARN")

1.2、创建RDD

在Spark中创建RDD的创建方式大概可以分为三种：

从集合中创建RDD从外部存储创建RDD从其他RDD创建

1.2.1、从集合中创建RDD

parallelize():通过parallelize函数把一般数据结构加载为RDDparallelize[T: ClassTag](seq: Seq[T],numSlices: Int = defaultParallelism): RDD[T]

Parallelize Rdd默认分区数：sc.defaultParallelism，可通过spark.default.parallelism设置sc.defaultParallelism的值，没有配置spark.default.parallelism时的默认值等于cpu的核数

例1

例2

1.2.2、从外部存储中创建RDD

通过textFile直接加载数据文件为RDD

textFile(path: String, minPartitions: Int = defaultMinPartitions): RDD[String]

读取HDFS时默认分区数：rdd的分区数 = max（hdfs文件的block数目， sc.defaultMinPartitions），sc.defaultMinPartitions=min(sc.defaultParallelism,2)

从本地文件读取：本地file的分片规则，应该按照hdfs的block大小划分，但实测的结果是固定按照32M来分片

读取HDFS上文件：

读取本地文件：

（注意：读取本地文件时，要确保每个集群上都要有文件，否则会报错）

任务1：

1、HDFS上有三份文件，分别为student.txt（学生信息表），result_bigdata.txt（大数据基础成绩表），result_math.txt（数学成绩表）

加载student.txt为名称为student的RDD数据，result_bigdata.txt为名称为bigdata的RDD数据，result_math.txt为名称为math的RDD数据

数据：

程序：

二、RDD算子

2.1、map与flatMap算子应用

2.1.1、map

map(func)

Transformation类型算子map: 将原来RDD的每个数据项通过map中的用户自定义函数f转换成一个新的RDD，map操作不会改变RDD的分区数目

示例：使用map函数对RDD中每个元素进行倍数操作

2.1.2、flatMap

flatMap(func)

Transformation类型算子flatMap：对集合中的每个元素进行map操作再扁平化

示例：使用flatMap分割单词

2.1.3、mapPartitions

mapPartitions(func)

Transformation类型算子和map功能类似，但是输入的元素是整个分区，即传入函数的操作对象是每个分区的Iterator集合，该操作不会导致Partitions数量的变化

示例：取出每个分区中大于3的值

2.2、sortBy与filter算子应用

2.2.1、sortBy

sortBy(f:(T) => K， ascending， numPartitions)

Transformation类型算子是可以对标准RDD进行排序sortBy()可接受三个参数：f:(T) => K：左边是要被排序对象中的每一个元素，右边返回的值是元素中要进行排序的值。ascending：决定排序后RDD中的元素是升序还是降序，默认是true，也就是升序，false为降序排序。numPartitions：该参数决定排序后的RDD的分区个数，默认排序后的分区个数和排序之前的个数相等。

示例：按照每个元素的第二个值进行降序排序

2.2.2、filter

filter(func)

Transformation类型算子保留通过函数func，返回值为true的元素，组成新的RDD过滤掉data RDD中元素小于或等于2的元素

示例：

任务2：

根据任务1得到的RDD bigdata及math，取出成绩排名前5的学生成绩信息

2.3、交集与并集计算的算子应用

2.3.1、distinct

distinct([numPartitions]))

Transformation类型算子针对RDD中重复的元素，只保留一个元素

示例：

2.3.2、union

union(otherDataset)

合并RDD，需要保证两个RDD元素类型一致

示例：合并rdd1和rdd2

2.3.3、intersection

intersection(otherDataset)

找出两个RDD的共同元素，也就是找出两个RDD的交集

示例：找出c_rdd1和c_rdd2中相同的元素

2.3.4、subtract

subtract (otherDataset)

获取两个RDD之间的差集

示例：找出rdd1与rdd2之间的差集

2.3.5、cartesian

cartesian(otherDataset)

笛卡尔积就是将两个集合的元素两两组合成一组

示例：

任务3：

1、找出考试成绩得过100分的学生ID，最终的结果需要集合到一个RDD中。

2、找出两门成绩都得100分的学生ID，结果汇总为一个RDD。

2.4、键值对RDD常用算子

虽然大部分Spark的RDD操作都支持所有种类的单值RDD，但是有少部分特殊的操作只能作用于键值对类型的RDD。

顾名思义，键值对RDD由一组组的键值对组成，这些RDD被称为PairRDD。PairRDD提供了并行操作各个键或跨节点重新进行数据分组的操作接口。例如，PairRDD提供了reduceByKey()方法，可以分别规约每个键对应的数据，还有join()方法，可以把两个RDD中键相同的元素组合在一起，合并为一个RDD。

2.4.1、创建键值对RDD

将一个普通的RDD转化为一个PairRDD时可以使用map函数来进行操作，传递的函数需要返回键值对。

做为键值对类型的RDD，包含了键跟值两个部分。Spark提供了两个方法分别获取键值对RDD的键跟值。keys返回一个仅包含键的RDD，values返回一个仅包含值的RDD。

2.4.2、mapValues

mapValues(func)

类似map，针对键值对（Key，Value）类型的数据中的Value进行map操作，而不对Key进行处理

示例：

2.4.3、groupByKey

groupByKey([numPartitions])

按键分组，在（K，V）对组成的RDD上调用时，返回（K，Iterable）对组成的新的RDD。

示例：将rdd按键进行分组

2.4.4、reduceByKey

将键值对RDD按键分组后进行聚合当在（K，V）类型的键值对组成的RDD上调用时，返回一个（K，V）类型键值对组成的新RDD其中新RDD每个键的值使用给定的reduce函数func进行聚合，该函数必须是（V，V）=>V类型

示例：统计每个键出现的次数

2.4.5、join

把键值对数据相同键的值整合起来其他连接有：leftOuterJoin, rightOuterJoin, and fullOuterJoin

join：把键值对数据相同键的值整合起来

2.5、常用Action类型算子

2.5.1、lookup

lookup(key: K)

Action类型算子作用于（K,V）类型的RDD上，返回指定K的所有V值

示例：

2.5.2、collect

collect()

返回RDD中所有的元素collectAsMap(): Map[K, V]

示例：

2.5.3、take

take(num)

返回RDD前面num条记录

示例：

2.5.4、count

count()

计算RDD中所有元素个数

任务4:

1、输出每位学生的总成绩，要求将两个成绩表中学生ID相同的成绩相加。

2、输出每位学生的平均成绩，要求将两个成绩表中学生ID相同的成绩相加并计算出平均分。

3、合并每个学生的总成绩和平均成绩。

三、文件读取与存储

3.1、saveAsTextFile

saveAsTextFile(path: String)

把RDD保存到HDFS中

3.2、repartition

repartition(numPartitions: Int)

可以增加或减少此RDD中的并行级别。在内部，它使用shuffle重新分发数据。如果要减少此RDD中的分区数，请考虑使用coalesce，这样可以避免执行shuffle。coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty)