创建你的第一个 PySpark DataFrame

介绍

零售商收到订单记录，希望将其转化为有用信息。计算销售报表之前，你需要一种方式来表示行、描述列，并请求处理工具执行工作。这里先使用三条能够直接目视核对的记录。

大数据指数据规模、到达速度或处理需求可能超出便利的单机方法所能应对的工作负载。Apache Spark 是能够将工作分配到集群多台机器上的数据处理引擎。PySpark 是它的 Python 接口。DataFrame 用具有名称和类型的列表示数据行，并支持筛选和聚合等操作。Spark 负责处理数据，独立的文件或存储系统则提供持久输入和输出。本课程先使用微小示例，让你先理解正确性，再考虑规模。

你将创建 SparkSession，显示 DataFrame，查看模式和行数，并增加一条记录后重新运行。预配置环境使用 local[2]：Spark 在当前机器上使用两个执行线程。这会教授真实 Spark API，分布式部署则是独立主题。你需要了解基本 Python 变量、列表和函数调用；Python 学习路径可帮助补充这些先修知识。无需 pandas、Hadoop、Hive、Java、安装软件或之前实验的文件。请使用终端。

运行完整的小型 DataFrame 脚本

在这一步中，你将运行完整脚本，把三条 Python 记录转为 Spark 表。先进入工作目录：

cd /home/labex/project

SparkSession 是创建 DataFrame 和访问 Spark 结构化数据 API 的入口。构建器设置应用名称、选择本地执行并创建会话。local[2] 允许在当前机器上使用两个执行线程，不会创建两台集群机器。setLogLevel 减少后续信息日志，让学习输出更容易找到。

Python 源码中的 from pyspark.sql import SparkSession 让程序能够使用已安装的这个类。点号访问指定成员，括号调用方法：df.show() 请求这个 DataFrame 显示行。构建器将多个调用串接起来，外围括号让表达式可以跨行。你可以将每行理解为一项设置，最后由 getOrCreate() 获取会话。

orders 列表包含三个元组，即括号包围的一组值。每个元组按声明列的顺序提供一行。模式字符串给每列命名并指定类型：文本标识符、商品，以及整数数量。createDataFrame 使用这些记录和模式。show 请求 Spark 显示行，stop 在完成时释放会话。

保存完整脚本。Shell 的 > 写入指定文件；带引号的 PYTHON 标记开始一个 here-document，原样保留其中的 Python 文本。请一直复制到最后的 PYTHON 行：

cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("FirstOrderTable")
    .master("local[2]")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")

orders = [
    ("s1", "notebook", 2),
    ("s2", "pen", 1),
    ("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)

df.show()

spark.stop()
PYTHON

预配置的 bigdata-python 命令选择已安装 PySpark 的 Python 环境。像普通 Python 程序一样运行文件，将标准输出保存到本地：

bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/first-run.txt

cat /home/labex/project/output/first-run.txt

预期看到列为 order_id、product、quantity 的表，包含 s1/notebook/2、s2/pen/1 和 s3/folder/3。启动警告或进度消息仍可能出现在终端中，它们与保存的表分开。没有明确排序时，不保证行顺序，因此应按订单 ID 辨认记录。

你的 Python 程序是 driver（驱动程序）：它创建会话，并请求 Spark 执行工作。在集群中，executor（执行器）进程针对数据片段运行任务。这里在同一台机器上本地执行。无需配置集群，你就有了可运行的程序。

检查模式并统计行数

在这一步中，你将区分列能够包含什么，以及表有多少条记录。模式（schema）列出列名、数据类型和是否允许空值。整数列表示适合数量计算的整数，文本则有不同的操作。允许空值表示字段可以缺失，不表示每一行都缺失该值。

添加 printSchema 查看描述，并用 count 计算行数。下面是完整更新文件，不依赖隐藏的交互状态：

cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("FirstOrderTable")
    .master("local[2]")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")

orders = [
    ("s1", "notebook", 2),
    ("s2", "pen", 1),
    ("s3", "folder", 3),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)

df.show()
df.printSchema()
print("Row count:", df.count())

spark.stop()
PYTHON

bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/schema-and-count.txt

cat /home/labex/project/output/schema-and-count.txt

表应该仍有三条记录。下方查找：

root
 |-- order_id: string (nullable = true)
 |-- product: string (nullable = true)
 |-- quantity: integer (nullable = true)

Row count: 3

printSchema 描述 DataFrame。show 和 count 是 action（行动操作），请求 Spark 生成可见结果或数值。Spark 可以推迟大部分处理，直到行动操作请求答案；后续实验将通过转换和计划解释这种惰性执行。计数 3 表示三条订单记录，不是数量总和，后者回答的是另一种业务问题。

每次调用都会启动新的 Python 进程并创建自己的 SparkSession。文件明确创建 df，不依赖交互式 Shell 曾经创建过 spark 或 df。

预测并测试新增记录

在这一步中，你将改变数据，同时保持其结构。预测增加 s4/notebook/1 后会改变什么：订单行数变为四，模式保持相同，原有三条记录保留。虽然 notebook 已在其他订单出现，新行仍然是一笔订单。

将脚本替换为只增加该条记录的这个版本：

cat > /home/labex/project/first_dataframe.py <<'PYTHON'
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("FirstOrderTable")
    .master("local[2]")
    .getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")

orders = [
    ("s1", "notebook", 2),
    ("s2", "pen", 1),
    ("s3", "folder", 3),
    ("s4", "notebook", 1),
]
columns = "order_id string, product string, quantity int"
df = spark.createDataFrame(orders, schema=columns)

df.show()
df.printSchema()
print("Row count:", df.count())

spark.stop()
PYTHON

在新进程中再次运行文件：

bigdata-python /home/labex/project/first_dataframe.py > /home/labex/project/output/expanded-run.txt

cat /home/labex/project/output/expanded-run.txt

找到全部四个订单 ID 和 Row count: 4。模式应该仍描述相同的三个列名和类型。重复商品名称不会自动将订单分组或删除重复记录；这些操作必须在后续课程中明确请求。

之前的 first-run.txt 和 schema-and-count.txt 仍保留三行版本的证据。可执行文件现在描述四行版本，因此另一次独立运行能够重现该结果。你已完成一轮小型的“修改—预测—运行—检查”，它将帮助你判断更大数据流程的正确性，而不是只相信程序成功退出。

总结

你运行了带有独立 SparkSession 的真实 PySpark 脚本，创建具有明确类型的 DataFrame，并检查显示记录、模式和行数。增加一条记录改变了行数，模式保持不变。接下来，你将从直接写在 Python 中的记录转向使用明确模式读取 CSV 文件。