PySpark 小白教程 (二) PySpark SQL 简介

PySpark SQL 简介

PySpark 是 Apache Spark 的 Python API，允许通过 Python 脚本进行大数据处理和分析。PySpark SQL 是 PySpark 的一个模块，用于处理结构化数据。它提供了与 SQL 语法相似的 DataFrame API，使数据工程师和数据科学家能够轻松地执行各种数据操作。

入门指南

要开始使用 PySpark SQL，您需要先安装 PySpark。可以通过以下命令安装：

pip install pyspark

安装完成后，可以创建一个 SparkSession，这是所有 Spark 功能的入口点：

from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder \
    .appName("PySpark SQL 基本操作") \
    .getOrCreate()

核心类

以下是 PySpark SQL 中的一些核心类：

• SparkSession: 提供了与 Spark 交互的入口点。
• DataFrame: 是分布式数据集，类似于关系数据库中的表。
• Column: 代表 DataFrame 中的列，用于表达数据操作。
• Row: 代表 DataFrame 中的行。
• GroupData: 表示分组后的数据，允许对分组数据进行聚合操作。

基本操作

1. 创建 DataFrame:
   可以从各种数据源（如 CSV、JSON、数据库等）创建 DataFrame：

   df = spark.read.csv("data.csv", header=True, inferSchema=True)
   df.show()
   

2. 选择列:
   使用 select 方法选择所需的列：

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

3. 过滤数据:
   使用 filter 方法进行数据过滤：

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

4. 分组聚合:
   使用 groupBy 和 agg 方法进行数据分组和聚合：

   df.groupBy("department").agg({"salary": "avg"}).show()
   

5. SQL 查询:
   可以直接在 DataFrame 上执行 SQL 查询：

   df.createOrReplaceTempView("employees")
   spark.sql("SELECT * FROM employees WHERE age > 30").show()
   

SparkSession 常用方法

• read: 读取数据源，生成 DataFrame。
• readStream: 读取流数据源，生成流 DataFrame。
• sql: 执行 SQL 查询。
• createDataFrame: 从 RDD、列表等创建 DataFrame。
• table: 访问临时或持久表。

DataFrame 常用方法

• show: 显示 DataFrame 的内容。
• select: 选择特定的列。
• filter: 过滤行。
• groupBy: 分组数据。
• agg: 聚合操作。
• join: 连接两个 DataFrame。

通过这些功能，PySpark SQL 提供了强大的工具来处理和分析大规模数据，使数据处理变得更加高效和灵活。