Python大数据之PySpark(五)RDD详解

本文涉及的产品
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
简介: Python大数据之PySpark(五)RDD详解

RDD详解

为什么需要RDD?

  • 首先Spark的提出为了解决MR的计算问题,诸如说迭代式计算,比如:机器学习或图计算
  • 希望能够提出一套基于内存的迭代式数据结构,引入RDD弹性分布式数据集,如下图
  • 为什么RDD是可以容错?
  • RDD依靠于依赖关系dependency relationship
  • reduceByKeyRDD-----mapRDD-----flatMapRDD
  • 另外缓存,广播变量,检查点机制等很多机制解决容错问题
  • 为什么RDD可以执行内存中计算?
  • RDD本身设计就是基于内存中迭代式计算
  • RDD是抽象的数据结构

什么是RDD?

  • RDD弹性分布式数据集
  • 弹性:可以基于内存存储也可以在磁盘中存储
  • 分布式:分布式存储(分区)和分布式计算
  • 数据集:数据的集合

RDD 定义

  • RDD是不可变,可分区,可并行计算的集合
  • 在pycharm中按两次shift可以查看源码,rdd.py
  • RDD提供了五大属性

RDD的5大特性

  • RDD五大特性:
  • 1-RDD是有一些列分区构成的,a list of partitions
  • 2-计算函数
  • 3-依赖关系,reduceByKey依赖于map依赖于flatMap
  • 4-(可选项)key-value的分区,对于key-value类型的数据默认分区是Hash分区,可以变更range分区等
  • 5-(可选项)位置优先性,移动计算不要移动存储
  • 1-
  • 2-
  • 3-
  • 4-
  • 5-最终图解
  • RDD五大属性总结
  • 1-分区列表
  • 2-计算函数
  • 3-依赖关系
  • 4-key-value的分区器
  • 5-位置优先性

RDD特点—不需要记忆

  • 分区
  • 只读
  • 依赖
  • 缓存
  • checkpoint

WordCount中RDD

RDD的创建

PySpark中RDD的创建两种方式

并行化方式创建RDD

rdd1=sc.paralleise([1,2,3,4,5])

通过文件创建RDD

rdd2=sc.textFile(“hdfs://node1:9820/pydata”)

代码:

# -*- coding: utf-8 -*-
# Program function:创建RDD的两种方式
'''
第一种方式:使用并行化集合,本质上就是将本地集合作为参数传递到sc.pa
第二种方式:使用sc.textFile方式读取外部文件系统,包括hdfs和本地文件系统
1-准备SparkContext的入口,申请资源
2-使用rdd创建的第一种方法
3-使用rdd创建的第二种方法
4-关闭SparkContext
'''
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
print("=========createRDD==============")
# 1 - 准备SparkContext的入口,申请资源
conf = SparkConf().setAppName("createRDD").setMaster("local[5]")
sc = SparkContext(conf=conf)
# 2 - 使用rdd创建的第一种方法
collection_rdd = sc.parallelize([1, 2, 3, 4, 5, 6])
print(collection_rdd.collect())  # [1, 2, 3, 4, 5, 6]
# 2-1 如何使用api获取rdd的分区个数
print("rdd numpartitions:{}".format(collection_rdd.getNumPartitions()))  # 5
# 3 - 使用rdd创建的第二种方法
file_rdd = sc.textFile("/export/data/pyspark_workspace/PySpark-SparkCore_3.1.2/data/words.txt")
print(file_rdd.collect())
print("rdd numpartitions:{}".format(file_rdd.getNumPartitions()))  # 2
# 4 - 关闭SparkContext
sc.stop()

文件读取

通过外部数据创建RDD


  • http://spark.apache.org/docs/latest/api/python/reference/pyspark.html#rdd-apis

# -*- coding: utf-8 -*-
# Program function:创建RDD的两种方式
'''
1-准备SparkContext的入口,申请资源
2-读取外部的文件使用sc.textFile和sc.wholeTextFile方式
3-关闭SparkContext
'''
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
 print("=========createRDD==============")
 # 1 - 准备SparkContext的入口,申请资源
 conf = SparkConf().setAppName("createRDD").setMaster("local[5]")
 sc = SparkContext(conf=conf)
 # 2 - 读取外部的文件使用sc.textFile和sc.wholeTextFile方式\
 file_rdd = sc.textFile("/export/data/pyspark_workspace/PySpark-SparkCore_3.1.2/data/ratings100")
 wholefile_rdd = sc.wholeTextFiles("/export/data/pyspark_workspace/PySpark-SparkCore_3.1.2/data/ratings100")
 print("file_rdd numpartitions:{}".format(file_rdd.getNumPartitions()))#file_rdd numpartitions:100
 print("wholefile_rdd numpartitions:{}".format(wholefile_rdd.getNumPartitions()))#wholefile_rdd numpartitions:2
 print(wholefile_rdd.take(1))# 路径,具体的值
 # 如何获取wholefile_rdd得到具体的值
 print(type(wholefile_rdd))#<class 'pyspark.rdd.RDD'>
 print(wholefile_rdd.map(lambda x: x[1]).take(1))
 # 3 - 关闭SparkContext
 sc.stop()
* 如何查看rdd的分区?getNumPartitions()

扩展阅读:RDD分区数如何确定



# -*- coding: utf-8 -*-
# Program function:创建RDD的两种方式
'''
第一种方式:使用并行化集合,本质上就是将本地集合作为参数传递到sc.pa
第二种方式:使用sc.textFile方式读取外部文件系统,包括hdfs和本地文件系统
1-准备SparkContext的入口,申请资源
2-使用rdd创建的第一种方法
3-使用rdd创建的第二种方法
4-关闭SparkContext
'''
from pyspark import SparkConf, SparkContext
if __name__ == '__main__':
print("=========createRDD==============")
# 1 - 准备SparkContext的入口,申请资源
conf = SparkConf().setAppName("createRDD").setMaster("local[*]")
# conf.set("spark.default.parallelism",10)#重写默认的并行度,10
sc = SparkContext(conf=conf)
# 2 - 使用rdd创建的第一种方法,
collection_rdd = sc.parallelize([1, 2, 3, 4, 5, 6],5)
# 2-1 如何使用api获取rdd的分区个数
print("rdd numpartitions:{}".format(collection_rdd.getNumPartitions()))  #2
# 总结:sparkconf设置的local[5](默认的并行度),sc.parallesise直接使用分区个数是5
# 如果设置spark.default.parallelism,默认并行度,sc.parallesise直接使用分区个数是10
# 优先级最高的是函数内部的第二个参数 3
# 2-2 如何打印每个分区的内容
print("per partition content:",collection_rdd.glom().collect())
# 3 - 使用rdd创建的第二种方法
# minPartitions最小的分区个数,最终有多少的分区个数,以实际打印为主
file_rdd = sc.textFile("/export/data/pyspark_workspace/PySpark-SparkCore_3.1.2/data/words.txt",10)
print("rdd numpartitions:{}".format(file_rdd.getNumPartitions()))
print(" file_rdd per partition content:",file_rdd.glom().collect())
# 如果sc.textFile读取的是文件夹中多个文件,这里的分区个数是以文件个数为主的,自己写的分区不起作用
# file_rdd = sc.textFile("/export/data/pyspark_workspace/PySpark-SparkCore_3.1.2/data/ratings100", 3)
# 4 - 关闭SparkContext
sc.stop()
* 首先明确,分区的个数,这里一切以看到的为主,特别在sc.textFile
  • 重要两个API
  • 分区个数getNumberPartitions
  • 分区内元素glom().collect()
相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
目录
相关文章
|
1月前
|
机器学习/深度学习 人工智能 分布式计算
我的阿里云社区年度总结报告:Python、人工智能与大数据领域的探索之旅
我的阿里云社区年度总结报告:Python、人工智能与大数据领域的探索之旅
114 35
|
30天前
|
人工智能 分布式计算 大数据
MaxFrame 产品评测:大数据与AI融合的Python分布式计算框架
MaxFrame是阿里云MaxCompute推出的自研Python分布式计算框架,支持大规模数据处理与AI应用。它提供类似Pandas的API,简化开发流程,并兼容多种机器学习库,加速模型训练前的数据准备。MaxFrame融合大数据和AI,提升效率、促进协作、增强创新能力。尽管初次配置稍显复杂,但其强大的功能集、性能优化及开放性使其成为现代企业与研究机构的理想选择。未来有望进一步简化使用门槛并加强社区建设。
68 7
|
1月前
|
SQL 分布式计算 DataWorks
MaxCompute MaxFrame评测 | 分布式Python计算服务MaxFrame(完整操作版)
在当今数字化迅猛发展的时代,数据信息的保存与分析对企业决策至关重要。MaxCompute MaxFrame是阿里云自研的分布式计算框架,支持Python编程接口、兼容Pandas接口并自动进行分布式计算。通过MaxCompute的海量计算资源,企业可以进行大规模数据处理、可视化数据分析及科学计算等任务。本文将详细介绍如何开通MaxCompute和DataWorks服务,并使用MaxFrame进行数据操作。包括创建项目、绑定数据源、编写PyODPS 3节点代码以及执行SQL查询等内容。最后,针对使用过程中遇到的问题提出反馈建议,帮助用户更好地理解和使用MaxFrame。
|
2月前
|
分布式计算 大数据 数据处理
技术评测:MaxCompute MaxFrame——阿里云自研分布式计算框架的Python编程接口
随着大数据和人工智能技术的发展,数据处理的需求日益增长。阿里云推出的MaxCompute MaxFrame(简称“MaxFrame”)是一个专为Python开发者设计的分布式计算框架,它不仅支持Python编程接口,还能直接利用MaxCompute的云原生大数据计算资源和服务。本文将通过一系列最佳实践测评,探讨MaxFrame在分布式Pandas处理以及大语言模型数据处理场景中的表现,并分析其在实际工作中的应用潜力。
106 2
|
3月前
|
并行计算 数据挖掘 大数据
Python数据分析实战:利用Pandas处理大数据集
Python数据分析实战:利用Pandas处理大数据集
|
4月前
|
分布式计算 Hadoop 大数据
大数据体系知识学习(一):PySpark和Hadoop环境的搭建与测试
这篇文章是关于大数据体系知识学习的,主要介绍了Apache Spark的基本概念、特点、组件,以及如何安装配置Java、PySpark和Hadoop环境。文章还提供了详细的安装步骤和测试代码,帮助读者搭建和测试大数据环境。
115 1
|
4月前
|
分布式计算 Java 大数据
大数据-92 Spark 集群 SparkRDD 原理 Standalone详解 ShuffleV1V2详解 RDD编程优化
大数据-92 Spark 集群 SparkRDD 原理 Standalone详解 ShuffleV1V2详解 RDD编程优化
60 0
大数据-92 Spark 集群 SparkRDD 原理 Standalone详解 ShuffleV1V2详解 RDD编程优化
|
4月前
|
消息中间件 分布式计算 Kafka
大数据-99 Spark 集群 Spark Streaming DStream 文件数据流、Socket、RDD队列流
大数据-99 Spark 集群 Spark Streaming DStream 文件数据流、Socket、RDD队列流
53 0
|
4月前
|
SQL 分布式计算 大数据
大数据-94 Spark 集群 SQL DataFrame & DataSet & RDD 创建与相互转换 SparkSQL
大数据-94 Spark 集群 SQL DataFrame & DataSet & RDD 创建与相互转换 SparkSQL
132 0
|
24天前
|
SQL 数据可视化 大数据
从数据小白到大数据达人:一步步成为数据分析专家
从数据小白到大数据达人:一步步成为数据分析专家
198 92