Spark RDD 转 Dataframe

TDHA • 2024-10-04 00:16 • 编程

一、关于SparkRDD

Apache Spark是一个开源的大数据计算框架，基于内存计算的方式提供了高效的数据处理能力。Spark运行于分布式集群上，利用RDD（Resilient Distributed Datasets）作为其计算模型。

下列说法正确的是：

1. RDD是Spark中的基本数据抽象，数据集通过分区的方式存储到分布式节点上，可以支持并行的数据操作。

2. RDD数据只能读写，不能进行更改。

3. RDD具有自动容错和恢复功能，即当某个节点出现故障时，Spark会重新计算丢失数据的节点，从而确保计算的完整性和正确性。

4. RDD的计算过程基于依赖关系（Lineage），以DAG（Directed Acyclic Graph）的方式进行数据处理操作，从而构造出了一个有向无环图。

二、Spark RDD转 Dataframe

Spark RDD提供了很多基本的数据操作函数，但是在进行数据处理时，我们往往会遇到一些问题。比如，RDD不支持动态数据类型，数据类型转换的效率较低，RDD的Schema不具有检查性等。这些问题可以通过使用Spark Dataframe来解决。

Spark Dataframe是一种基于RDD的分布式数据结构，它类似于传统的关系型数据库，具有Schema、列和行等概念。相比于RDD，Dataframe具有以下优势：

1. 支持动态数据类型，因此可以使用基于数据类型的操作例如过滤、排序、聚合和统计。

2. 支持数据集编码和解码，因此可以有效地处理Python和Java对象。

3. 提供了API最大化的优化和调优，因此具有更好的性能。

4. 提供了Spark SQL引擎，因此可以通过SQL和HiveQL查询Dataframe。

我们可以通过Spark SQL中的API，将RDD转换为Dataframe。

# 引入pyspark包中的SparkSession
from pyspark.sql import SparkSession

# 创建SparkSession对象
spark = SparkSession.builder.appName("RDD to DataFrame").getOrCreate()

# 创建RDD
rdd = spark.sparkContext.parallelize([(1, "John"), (2, "Doe"), (3, "Lena"), (4, "Joe")])

# RDD转换为Dataframe
df = rdd.toDF(["id", "name"])

# 显示Dataframe
df.show()

上面的代码首先创建了一个SparkSession对象，然后创建了一个RDD，最后将RDD转换为Dataframe，并显示出来。其中“toDF()”方法将RDD转换为Dataframe，参数“[‘id’, ‘name’]”是Dataframe的列名称。

三、示例

1. 从CSV文件创建Dataframe

首先，我们需要加载CSV文件并创建一个RDD。

# 引入pyspark包中的SparkSession
from pyspark.sql import SparkSession

# 创建SparkSession对象
spark = SparkSession.builder.appName("CSV to DataFrame").getOrCreate()

# 从CSV文件创建RDD
rdd = spark.sparkContext.textFile("file.csv").map(lambda line: line.split(","))

# RDD转换为Dataframe
df = rdd.toDF(["col1", "col2", "col3"])

# 显示Dataframe
df.show()

上面的代码加载了名为“file.csv”的CSV文件，并将其转换为RDD。然后，将RDD转换为Dataframe，并将其列名称设置为“col1”、“col2”、“col3”。

2. Dataframe中的列操作

我们可以对Dataframe中的列进行各种操作，比如选择某些列、添加新列、删除列、更改列名等。

选择某些列

# 选择id和name列
df.select("id", "name")

添加新列

# 添加age列
from pyspark.sql.functions import lit
df.withColumn("age", lit(25))

删除列

# 删除name列
df.drop("name")

更改列名

# 将name列更改为first_name
df.withColumnRenamed("name", "first_name")

参考资料

Apache Spark官方文档：https://spark.apache.org/docs/latest/
Spark SQL官方文档：https://spark.apache.org/docs/latest/sql-programming-guide.html

原创文章，作者：TDHA，如若转载，请注明出处：https://www.506064.com/n/136915.html