跳到主要内容

Spark DataFrame API

介绍​

Spark DataFrame API 是 Apache Spark 中用于处理结构化数据的核心 API 之一。它提供了一个高级抽象，允许用户以类似于 SQL 或 Pandas DataFrame 的方式操作数据。DataFrame 是一个分布式的数据集合，具有命名的列，类似于关系数据库中的表。与 RDD（弹性分布式数据集）相比，DataFrame 提供了更高效的执行计划和更丰富的优化功能。

DataFrame API 的主要优势包括：

  • 易用性：支持多种编程语言（如 Scala、Java、Python 和 R）。
  • 高性能：通过 Catalyst 优化器和 Tungsten 执行引擎实现高效的数据处理。
  • 丰富的功能：支持 SQL 查询、数据聚合、过滤、连接等操作。

创建 DataFrame​

在 Spark 中，可以通过多种方式创建 DataFrame。以下是一个从 CSV 文件创建 DataFrame 的示例：

python
from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder.appName("DataFrameExample").getOrCreate()

# 从 CSV 文件创建 DataFrame
df = spark.read.csv("data.csv", header=True, inferSchema=True)

# 显示 DataFrame 的前 5 行
df.show(5)

输入：假设 data.csv 文件包含以下内容：

name,age,city
Alice,30,New York
Bob,25,San Francisco
Cathy,35,Los Angeles

输出：

+-----+---+-------------+
| name|age|         city|
+-----+---+-------------+
|Alice| 30|     New York|
|  Bob| 25|San Francisco|
|Cathy| 35|  Los Angeles|
+-----+---+-------------+

基本操作​

选择列​

可以使用 select 方法选择 DataFrame 中的特定列：

python
df.select("name", "age").show()

输出：

+-----+---+
| name|age|
+-----+---+
|Alice| 30|
|  Bob| 25|
|Cathy| 35|
+-----+---+

过滤数据​

使用 filter 方法可以根据条件过滤数据：

python
df.filter(df["age"] > 30).show()

输出：

+-----+---+------------+
| name|age|        city|
+-----+---+------------+
|Cathy| 35| Los Angeles|
+-----+---+------------+

数据聚合​

可以使用 groupBy 和 agg 方法进行数据聚合：

python
df.groupBy("city").agg({"age": "avg"}).show()

输出：

+-------------+--------+
|         city|avg(age)|
+-------------+--------+
|     New York|    30.0|
|San Francisco|    25.0|
|  Los Angeles|    35.0|
+-------------+--------+

实际应用场景​

案例：分析销售数据​

假设我们有一个销售数据的 CSV 文件 sales.csv，包含以下字段：product, quantity, price, date。我们可以使用 DataFrame API 来分析这些数据。

python
# 读取销售数据
sales_df = spark.read.csv("sales.csv", header=True, inferSchema=True)

# 计算每个产品的总销售额
sales_df.withColumn("total_sales", sales_df["quantity"] * sales_df["price"]) \
        .groupBy("product") \
        .agg({"total_sales": "sum"}) \
        .show()

输出：

+-------+-------------+
|product|sum(total_sales)|
+-------+-------------+
|  Apple|        150.0|
| Banana|         75.0|
| Orange|        120.0|
+-------+-------------+

总结​

Spark DataFrame API 提供了一种高效且易于使用的方式来处理大规模结构化数据。通过本文，你已经学习了如何创建 DataFrame、进行基本操作以及如何在实际场景中应用这些知识。希望这些内容能够帮助你更好地理解和使用 Spark DataFrame API。

附加资源​

提示

练习：尝试使用 Spark DataFrame API 分析你感兴趣的数据集，并分享你的分析结果！