在 PySpark 中选列、筛选并计算订单金额

介绍

类型正确的订单表只是起点，还不是答案。同事可能只需要几个列、已付款订单，以及由数量和单价算出的金额。你将用 Spark DataFrame 操作表达这些选择，同时保留原始记录。

列表达式（column expression）描述要让 Spark 应用到行上的计算或条件。你将选择字段、组合条件、增加十进制金额，并给结果列设置易读名称。这些操作创建新的 DataFrame；show 等行动操作则请求实际结果。这个区别能帮助解释 Spark 惰性执行，暂时无需了解调度内部机制。

本实验独立提供六个干净零售订单。你应该理解前一个实验中的明确 CSV 模式和基本 Python。每个可运行版本都包含完整读取器和 SparkSession，因此无需之前的进程或文件。请使用终端和预配置 bigdata-python 环境。

选择问题所需的列

在这一步中，你将缩减表的列，同时保留全部六个订单记录。先查看源数据：

cd /home/labex/project

cat /home/labex/project/data/orders.csv

select 选择新 DataFrame 呈现哪些列，以及它们的顺序，不筛选行。下面的完整脚本重复熟悉的明确读取器，然后选择订单标识符、数量、价格和状态。虽然 selected 不呈现日期和客户字段，它们仍保留在 df 中。

用带引号的 here-document 保存并运行脚本。> 写入文件或替换先前版本；最后的 PYTHON 标记结束源码文本：

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

selected = df.select("order_id", "quantity", "unit_price", "order_status")
selected.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/selected-orders.txt

cat /home/labex/project/output/selected-orders.txt

预期显示全部六个订单 ID 和四列。orderBy 让显示可预测，show 触发返回行所需的工作。这是列选择，也称为投影（projection）。没有源文件改变，也没有订单被移除。

组合条件选择行

在这一步中，你将保留数量至少为二的已付款订单。col("order_status") 引用 Spark 列；用 == "paid" 比较会构建 Spark 条件，不是一个 Python True 或 False。>= 构建数量条件。

用 & 以 AND 语义组合 Spark 列条件。每个比较外都要加括号，因为 Python 运算符优先级会影响结果。普通 Python and 不是组合 Spark Column 对象的运算符。

预测符合条件的 ID：o1、o4、o6 同时满足两个条件。o2 已付款，但数量只有一。替换并运行完整脚本：

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

eligible = df.filter((col("order_status") == "paid") & (col("quantity") >= 2))
eligible.select("order_id", "quantity", "unit_price", "order_status").orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/eligible-orders.txt

cat /home/labex/project/output/eligible-orders.txt

结果应该只有这三个 ID，每个数量为 2。filter 改变所选行，select 控制显示列。你可以组合两种操作，而 df 继续描述完整的六行源数据。

本步数量门槛只是演示条件，不是销售的通用定义。下一项计算将包含全部四个已付款订单，包括 o2，因为每个已付款订单都贡献收入。

计算并命名十进制订单金额

在这一步中，你将用数量乘以十进制单价，计算每个已付款订单的金额。withColumn 向新 DataFrame 添加命名列；如果同名列已存在，则替换该列。这里的 order_amount 使用 Spark 内置乘法表达式计算。

接下来的 select 保留订单 ID，并用 alias 将计算列的显示名设为 amount。别名改变结果列名，不编辑 CSV 表头。全部四个已付款订单都应参与计算；待付款和已退款行被排除，在这里的简化规则中不把退款当作负收入。

运行完整版本：

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/paid-amounts.txt

cat /home/labex/project/output/paid-amounts.txt

预期为 o1/20.00、o2/25.00、o4/50.00 和 o6/30.00，可能带额外末尾零。模式应将 amount 标识为十进制类型。Spark 在乘法中可能扩大十进制精度；重点是保留十进制运算，没有将价格转成二进制浮点数。

filter、withColumn、select 和 orderBy 描述转换（transformation）并返回新 DataFrame。定义这条操作链不会立即把每个中间表都计算并存储出来。show 是行动操作，请求 Spark 执行所需工作并显示答案。printSchema 描述列结构。之后你将检查这类操作链背后的实际执行计划。

改变一个状态并预测影响

在这一步中，你将保留已付款结果，再添加独立的待付款订单练习视图。先预测输出：只有 o3 待付款，数量一、价格 10.00，因此金额为 10.00。

新的 pending_result 使用相同表达式链，只改变状态值。它从 df 开始，不是从已筛选的 paid_result 开始，所以待付款记录仍可访问：

cat > /home/labex/project/transform_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType

spark = SparkSession.builder.appName("TransformOrders").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
    StructField("order_id", StringType(), True),
    StructField("customer_id", StringType(), True),
    StructField("product_id", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("order_status", StringType(), True),
    StructField("order_date", DateType(), True),
])
df = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")

paid_result = (
    df.filter(col("order_status") == "paid")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Paid orders:")
paid_result.orderBy("order_id").show(truncate=False)
paid_result.printSchema()

pending_result = (
    df.filter(col("order_status") == "pending")
    .withColumn("order_amount", col("quantity") * col("unit_price"))
    .select("order_id", col("order_amount").alias("amount"))
)
print("Pending practice:")
pending_result.orderBy("order_id").show(truncate=False)

spark.stop()
PYTHON

bigdata-python /home/labex/project/transform_orders.py > /home/labex/project/output/status-comparison.txt

cat /home/labex/project/output/status-comparison.txt

Paid orders 下应保留相同的四个已付款金额。Pending practice 下应只有 o3/10.00。如果对已付款视图再筛选待付款状态，将返回零行，因为那些记录已被排除；从合适的源开始很重要。

两条操作链都由 Spark 内置表达式计算。你没有把一般数据集收集到 Python 列表后再循环处理。这让处理描述适用于 Spark 执行引擎。最终脚本保留已付款计算和小型练习比较，原始 CSV 则仍含全部六个订单。

总结

你选择了列，用 Spark 表达式组合行条件，计算十进制订单金额，并命名输出列。比较已付款与待付款视图，展示了转换为什么应从合适的源开始，以及行动操作如何请求结果。接下来，你将把熟悉的读取和选择方法应用到 JSON 输入。

选择、筛选并转换 Spark 订单数据

介绍

选择问题所需的列

组合条件选择行

计算并命名十进制订单金额

改变一个状态并预测影响

总结