RDD的创建方式(Python)_创建一个包含1到10的整数的rdd

一、先说说 RDD、DataFrame和 Dataset 的区别

1.RDD、DataFrame、Dataset 概念：

Spark RDD： RDD代表弹性分布式数据集。它是记录的只读分区集合。 RDD是Spark的基本数据结构。它允许程序员以容错方式在大型集群上执行内存计算。
Spark Dataframe： 与RDD不同，数据以列的形式组织起来，类似于关系数据库中的表。它是一个不可变的分布式数据集合。 Spark中的DataFrame允许开发人员将数据结构(类型)加到分布式数据集合上，从而实现更高级别的抽象，DataFrame 与 RDD 的主要区别在于，前者带有 schema 元信息，即 DataFrame 所表示的二维表数据集的每一列都带有名称和类型。。
Spark Dataset： Apache Spark中的Dataset是DataFrame API的扩展，它提供了类型安全(type-safe)，面向对象(object-oriented)的编程接口。 Dataset利用Catalyst optimizer可以让用户通过类似于sql的表达式对数据进行查询。DataSet在DataFrame基础上支持更强的数据类型控制，更精细化操作数据，避免因数据类型异常，造成的程序运行异常。
DataSet是Spark1.6添加的分布式数据集合，Spark2.0合并DataSet和DataFrame数据集合API，DataFrame变成DataSet的子集。
DataSet继承RDD优点，并使用Spark SQL优化的执行引擎。支持JVM对象构建，支持函数式转换（map/flatmap/filter）等多种操作

2.RDD、DataFrame和 Dataset 三者的共性：

1. RDD、DataFrame、DataSet全都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利;
2. 三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action如foreach时，三者才会开始遍历运算;
3. 三者有许多共同的函数，如filter，排序等;
4. 在对DataFrame和Dataset进行操作许多操作都需要这个包:import spark.implicits._（在创建好SparkSession对象后尽量直接导入）；
5. 三者都会根据 Spark 的内存情况自动缓存运算，这样即使数据量很大，也不用担心会内存溢出；
6. 三者都有partition的概念；
7. DataFrame和Dataset均可使用模式匹配获取各个字段的值和类型。

二、创建RDD的方式

在PySpark中，可以使用以下方式创建RDD：

1. 从内存中创建RDD：使用SparkContext的parallelize()方法可以将一个Python列表或元组转换为RDD。
2. 从外部存储系统中创建RDD：可以通过textFile()方法从HDFS、本地文件系统、S3等存储系统中读取数据，创建RDD。
3. 通过转换操作（转换算子）创建RDD：可以通过对已有的RDD进行转换操作（转换算子），创建新的RDD。

从内存、外部存储系统和转换操作（转换算子）三种方式创建RDD的方法。需要注意的是，创建RDD需要先创建SparkContext对象，以便与Spark集群进行通信。

二、代码说明

1. 从内存中创建RDD

from pyspark import SparkContext
sc = SparkContext("local", "RDD Creation Example")
# 创建一个包含1到10整数的列表
data = range(1, 11)
# 将列表转换为RDD
rdd = sc.parallelize(data)
# 输出RDD的元素
print(rdd.collect())
1
2
3
4
5
6
7
8

2. 从外部存储系统中创建RDD

from pyspark import SparkContext
sc = SparkContext("local", "RDD Creation Example")
# 从本地文件系统中读取数据，创建RDD
rdd = sc.textFile("file:///path/to/file.txt")
# 输出RDD的元素
print(rdd.collect())
1
2
3
4
5
6

3. 通过转换操作（转换算子）创建RDD

from pyspark import SparkContext
sc = SparkContext("local", "RDD Creation Example")
# 创建一个包含1到10整数的列表
data = range(1, 11)
# 将列表转换为RDD
rdd1 = sc.parallelize(data)
# 将rdd1中的元素乘以2，创建新的RDD
rdd2 = rdd1.map(lambda x: x * 2)
# 输出新的RDD的元素
print(rdd2.collect())
1
2
3
4
5
6
7
8
9
10