介绍
零售商的订单记录现在以 CSV 文件交付，不再直接写在 Python 列表里。文件保存 2、10.00 这样的文本，而计算需要知道这些值表示文本、整数数量，还是十进制价格。读到正确字符，只是正确读取数据的一部分。
你将比较 Spark 默认的字符串读取、自动类型推断和明确模式。然后，对字段顺序相同的第二个文件应用相同模式。这里的业务数据特意保持有效，让你专注于列和值之间的约定。之后的实验将明确处理缺失和无效输入。
环境提供 PySpark 和两个独立本地 CSV 样本。你应该理解第一个实验中的 SparkSession、DataFrame 显示、模式和行数，以及基本 Python。无需 pandas、Hadoop、Hive、安装软件或之前的虚拟机状态。请使用终端。
读取表头并观察字符串列
在这一步中，你将把 CSV 字段位置与 Spark 创建的表联系起来。先查看实际文件：
cd /home/labex/project
cat /home/labex/project/data/orders.csv
第一行给七列命名，接下来的六行是订单。每行表示一个只含一种商品类型的订单。文件包含 paid、pending 和 refunded 状态；本实验全部读取，不应用销售规则。
spark.read 创建 DataFrame 读取器。option("header", True) 告诉读取器将 CSV 第一行视为列名，而不是订单记录。csv(input_path) 指定本地文件。如果没有类型推断或提供的模式，CSV 读取器会生成字符串列。
保存并运行完整程序。带引号的 PYTHON here-document 原样保留其中源码，> 写入本地文件。bigdata-python 使用预配置的 PySpark 环境：
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/orders.csv"
df = spark.read.option("header", True).csv(input_path)
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/string-orders.txt
cat /home/labex/project/output/string-orders.txt
预期有六行、七个字符串列，包括 quantity 和 unit_price。显示的 2 可能看起来像数字，但 printSchema 指出了它实际为字符串。orderBy("order_id") 按标识符排序显示行，让检查可预测；排序与类型指定是独立操作。
header=True 避免表头成为第七条记录，却没有告诉 Spark 价格应该是精确十进制数，或日期应该支持日期操作。
比较推断与明确的数据约定
在这一步中，你将声明每个 CSV 字段的含义。inferSchema=True 请求 Spark 检查值并推断类型。这很方便，但结果取决于观察到的数据，不是对价格或后续文件的持久约定。在这个样本中，推断将 unit_price 识别为浮点数 double。
用明确模式表达预期约定。StructType 是有序字段集合。每个 StructField 包含名称、类型，以及允许缺失值的 True。StringType 保存标识符和状态文本，IntegerType 保存数量，DecimalType(10, 2) 保存总计十位、其中两位小数的精确价格，DateType 保存日历日期。允许空值并不意味着业务信息缺失就自动可以接受。
将程序替换为以下版本。它先打印推断类型描述供比较，再使用明确模式读取相同文件：
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").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),
])
input_path = "/home/labex/project/data/orders.csv"
inferred = spark.read.option("header", True).option("inferSchema", True).csv(input_path)
print("Inferred schema:", inferred.schema.simpleString())
df = (
spark.read
.option("header", True)
.option("dateFormat", "yyyy-MM-dd")
.schema(order_schema)
.csv(input_path)
)
print("Explicit schema:")
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/typed-orders.txt
cat /home/labex/project/output/typed-orders.txt
在推断汇总下找到明确模式。quantity 应为 integer，unit_price 为 decimal(10,2)，order_date 为 date。全部六条记录仍应存在。明确的 dateFormat 选项将输入日期描述为年-月-日，与 2026-10-14 这样的值一致。
模式字段顺序与这个 CSV 的字段顺序相同。提供模式不意味着会根据表头名称自动重排所有输入文件。对于这个顺序已知的交付，约定明确表达了预期列和类型。模式也不能替代数据质量策略；之后你将学习如何安全处理无效值。
对新收到的文件应用相同模式
在这一步中，你将用第二份有效文件测试约定。先查看它：
cat /home/labex/project/data/new-orders.csv
它有顺序相同的七个字段，包含两个新订单。预测变化：业务行数从六变为二，数量、价格和日期则保留声明的类型。
下面的完整版本保留明确模式，改变输入路径。正常读取时不再需要推断比较：
cat > /home/labex/project/read_orders.py <<'PYTHON'
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("ReadOrderCSV").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),
])
input_path = "/home/labex/project/data/new-orders.csv"
df = (
spark.read
.option("header", True)
.option("dateFormat", "yyyy-MM-dd")
.schema(order_schema)
.csv(input_path)
)
print("Explicit schema:")
df.printSchema()
df.orderBy("order_id").show(truncate=False)
print("Rows:", df.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_orders.py > /home/labex/project/output/new-typed-orders.txt
cat /home/labex/project/output/new-typed-orders.txt
预期 o7 数量 2、价格 12.50、状态 paid、日期 2026-10-16；o8 数量 3、价格 4.00、状态 pending、日期 2026-10-17。计数应为 2。检查 printSchema 中相同的整数、十进制和日期类型。
原来的六订单文件保持不变。可复用模式让新交付文件进入相同处理逻辑，而不必重新定义数量或价格的含义。接下来，你将使用这些有类型的列选择订单并计算每笔金额。
总结
你正确读取了 CSV 表头，比较了字符串读取和类型推断，并为标识符、数量、十进制价格和日期声明明确约定。相同模式正确解释了第二批数据。接下来，有类型的列表达式将让你筛选行并计算订单金额。
