介绍
零售商已经有一组正确、具有明确类型的已付款订单结果。现在，另一个应用需要稍后读取这些结果、选择某个日期，并确信重复导出没有悄悄复制订单。保存数据是处理约定的一部分，不只是脚本的最后一行。
你将使用按 order_date 的目录分区写入 Parquet。数据集根目录中包含按日期命名的子目录，Spark 读取根目录时会重建日期列。这些存储目录与执行工作使用的计算分区不同；下个实验将单独检查执行。
本实验独立在本地 Parquet 输入中准备四笔已付款订单及其十进制金额。你应该了解 DataFrame 读取、筛选和最小覆盖写 Parquet 流程。无需 Hive 目录服务、HDFS、SQL 基础、安装软件或之前的输出。请使用终端。
写入日期分区数据集并读回
在这一步中，你将把预置结果组织为日期目录。data/prepared-orders 下的输入数据集包含四个订单，使用与之前相同的有类型数量、十进制价格、日期和十进制 order_amount。Parquet 是二进制格式，请通过 Spark 检查，不要期望 cat 显示 CSV 文本。
进入工作目录：
cd /home/labex/project
partitionBy("order_date") 告诉写入器将行放入按日期命名的目录。mode("overwrite") 替换指定输出路径中已有的派生数据集。该输出路径应与必须保持完整的 input_path 分开。
保存并执行完整程序：
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/partitioned-read-back.txt
cat /home/labex/project/output/partitioned-read-back.txt
预期重建四条记录：o1、o2 位于 10 月 14 日，o4、o6 位于 10 月 15 日，金额分别为 20.00、25.00、50.00、30.00。Main rows 应为 4。检查 quantity 为整数、价格和金额为十进制、order_date 为日期。
分区列读回后可能出现在不同位置，因为 Spark 从目录名重建它，不把它作为每个分区文件中的普通字段存储。列名和值确定身份；下游使用者要求特定列顺序时，应明确 select。显示顺序也需要 orderBy。
检查目录并选择某一天
在这一步中，你将把目录布局与重建的日期列联系起来。递归列出输出；-R 包含日期子目录及其中的文件：
ls -R /home/labex/project/output/orders-by-date
预期有 order_date=2026-10-14 和 order_date=2026-10-15 目录。part 文件名、数量及隐藏元数据可能变化；不要把某个 part 文件硬编码为整个数据集。有意义的根目录是 orders-by-date。
读取根目录让 Spark 发现分区值。这种文件布局发现不要求注册 Hive 分区或表。下面独立脚本读取根目录，将 DateType 列与有效 ISO 日期字符串比较，并单独保存选定日期的记录：
cat > /home/labex/project/inspect_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
spark = SparkSession.builder.appName("InspectDateDataset").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
root = "/home/labex/project/output/orders-by-date"
df = spark.read.parquet(root)
selected = df.filter(col("order_date") == "2026-10-14")
selected.orderBy("order_id").show(truncate=False)
selected.write.mode("overwrite").parquet("/home/labex/project/output/selected-day")
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_sales_data.py > /home/labex/project/output/selected-date.txt
cat /home/labex/project/output/selected-date.txt
预期只有 o1 和 o2，日期均为 10 月 14 日，金额为 20.00、25.00。主数据集中另一个日期仍然存在。筛选读取改变返回记录，不删除未选择目录。
你检查了存储组织和实际查询结果。对这样的小样本，耗时不能证明某种格式或布局更快。目的是生成正确、可发现、供后续读取器使用的数据集。
安全比较重复覆盖与追加
在这一步中，你将保留正确主数据集，在独立目录中实验。预测临时目录的计数：把相同四行覆盖写两次，每次仍为四行；之后追加完整副本则生成八行。额外行是重复导出，不是新订单。
运行完整版本。主路径使用覆盖，只有 write-mode-demo 接受有意的追加：
cat > /home/labex/project/write_sales_data.py <<'PYTHON'
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("WriteSalesData").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
input_path = "/home/labex/project/data/prepared-orders"
output_path = "/home/labex/project/output/orders-by-date"
prepared = spark.read.parquet(input_path)
prepared.write.mode("overwrite").partitionBy("order_date").parquet(output_path)
print("Main read-back:")
reloaded = spark.read.parquet(output_path)
reloaded.orderBy("order_id").show(truncate=False)
reloaded.printSchema()
print("Main rows:", reloaded.count())
practice_path = "/home/labex/project/output/write-mode-demo"
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("First overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("overwrite").partitionBy("order_date").parquet(practice_path)
print("Second overwrite rows:", spark.read.parquet(practice_path).count())
prepared.write.mode("append").partitionBy("order_date").parquet(practice_path)
print("After append rows:", spark.read.parquet(practice_path).count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
找到 Main rows: 4、First overwrite rows: 4、Second overwrite rows: 4 和 After append rows: 8。临时数据集现在有意包含每笔订单的两份副本，展示了为什么追加完整快照不适合主导出。
再次运行保存的程序，检查整个流程的可重复性：
bigdata-python /home/labex/project/write_sales_data.py > /home/labex/project/output/write-mode-comparison.txt
cat /home/labex/project/output/write-mode-comparison.txt
应该出现相同计数。每次完整运行都重建主数据集，并在有意追加前重置临时实验，因此不会增长为十二或十六行。追加是针对真正新增且不重叠数据的另一种策略；本次完整派生快照选择覆盖。
原始 prepared-orders 数据集仍然可用。日期目录描述存储值，Spark 运行时计算分区则描述作业如何拆分工作。接下来，你将把分区的第二层含义与计划和实际执行联系起来。
总结
你写入了日期分区 Parquet 数据集，从根目录重建有类型记录，并选择某一天而没有移除其他日期。重复覆盖保留主快照，临时追加则展示了重复副本。接下来，执行计划和真实事件将解释计算分区、作业、阶段和任务。
