Spark笔记6-RDD创建和操作
Spark笔记6-RDD创建和操作
皮大大
发布于 2021-03-02 15:32:04
发布于 2021-03-02 15:32:04
代码可运行
运行总次数：0
代码可运行
RDD创建
从文件系统中加载数据生成RDD
spark的sparkcontext通过textfile()读取数据生成内存中的RDD，文件来源：
- 本地文件系统
- 分布式文件系统
HDFS Amazon S3等云端的文件
# 本地加载
lines = sc.textFile("file:///usr/local/spark/rdd/word.txt")
lines.foreach(print) # 查看具体信息
# 分布式文件系统进行加载数据
lines = sc.textFile("hdfs://localhost:9000/usr/hadoop/word.txt")
lines = sc.textFile("/usr/hadoop/word.txt") # 去掉绝对路径
lines = sc.textFile("word.txt") # 文件放在当前主用户的目录下通过并行集合（数组）创建RDD的实例
并行parallelize()方法创建
array = [1,3,4,5,2]
rdd = sc.parallelize(array)
rdd.foreach(print)RDD操作
转换操作
RDD是只读的，只能在修改的过程进行修改。转换过程是惰性机制。整个转换过程只是记录转换的轨迹，并不会发生真正的计算。只有遇到行动操作action时候，才会发生真正的计算。三种操作：
- filter
- map
- flatmap
- groupbykey
filter(func)
筛选满足函数func的元素，并且返回一个新的数据集
lines = sc.textFile("word.txt")
linesWithSpark = lines.filter(lambda line: "Spark" in line)
lineWithSpark.foreach(print)map(func)
将RDD对象中的元素放入func函数中进行操作
data = [1,2,3,4]
rdd1 = sc.parallelize(data)
rdd2 = rdd1.map(lambda x: x+10)
rdd2.foreach(print)
lines = sc.textFile("word.txt")
words = lines.map(lambda line:line.split(" ")) # 函数功能是用空格进行分割
words.foreach(print)
flatmap(func)
与map比较类似，但是每个输入元素都可以映射到0个或者多个输出结果（可乐罐栗子）
lines = sc.textFile("file:///usr/local/spark/code/radd/word.txt") # 加载本地文件
words = lines.flatMap(lambda line:line.split(" "))
groupbykey()
通过key进行分组；相同的值放到一个元组中，是以iterable的数据形式存放。
words = sc.paralelize(obj)
words1 = words.groupByKey()
wordsq.foreach(print)
reduceByKey(func)
将返回的值value通过func函数进行计算
行动操作action
执行行动类型操作，发生真正的计算
函数 | 说明 |
|---|---|
count() | 返回数据集中的总个数 |
collect() | 以列表或数组的形式返回数据集中的所有元素 |
first() | 返回第一个元素 |
take(n) | 以列表的形式返回前n个元素 |
reduce(func) | 通过func函数聚合数据集中的所有元素 |
foreach(func) | 将数据集中的元素传递给函数func进行运行 |
惰性机制
在RDD的操作中，只有遇到行动类型的操作才是开始计算操作
lines = sc.textFile("word.txt")
linelength = lines.map(lambda s: len(s)) # 记录轨迹
totallength = linelength.reduce(lambda a,b: a+b) # 遇到reduce函数才开始执行计算过程
print(totallength)本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。
原始发表：2019-10-30，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
腾讯云开发者
