介绍
有些系统以 CSV 行交付记录，另一些则交付包含命名字段的 JSON 对象。零售商现在提供事件文件，其中有类型标签和订单相关值。处理问题依然熟悉：字段表示什么，哪些记录属于订单，结果应该呈现哪些列？
JSON Lines 每行保存一个完整 JSON 对象。与 CSV 不同，字段名随每个对象一起保存，对象中也可以包含另一个对象。你将先使用明确类型读取平面记录，再从一个简单嵌套客户对象中选择字段，最后将模式应用到第二批数据。
这些是用 spark.read 读取的有限文件，因此仍然是批处理。文件名包含 events，并不会让应用变成流处理。环境和数据均独立准备；你需要前面的明确模式、select、filter、alias 概念和基本 Python。无需 Kafka、Hadoop、Hive、流处理配置、数组或复杂递归。请使用终端。
读取平面 JSON Lines 并选择订单事件
在这一步中，你将把明确模式读取方法从 CSV 迁移到 JSON。查看提供的文件：
cd /home/labex/project
cat /home/labex/project/data/flat-events.jsonl
每行是由花括号包围的对象。双引号包围的名称标识字段，冒号分隔名称和值。quantity 等数字是 JSON 数值，order_date 则是带引号的日期字符串，将由声明的 DateType 解释。这里没有 CSV 表头，也没有外围数组。
第一个对象的 order_id、数量、价格和日期，描述与 CSV 订单相同类型的值。文件也包含 inventory（库存）事件；记录有效可读，不代表它属于订单结果。下面的明确模式声明平面字段。json(path) 读取该格式，熟悉的 filter 选择 event_type == order。
用带引号的 here-document 保存完整脚本，并通过预配置的 PySpark 解释器运行：
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/flat-events.jsonl")
)
events.printSchema()
events.orderBy("order_id").show(truncate=False)
orders = events.filter(col("event_type") == "order")
print("Order events:")
orders.select("order_id", "quantity", "unit_price", "order_date").orderBy("order_id").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/flat-json-orders.txt
cat /home/labex/project/output/flat-json-orders.txt
初始显示应包含全部三个事件记录。Order events 下应只有 j1/2/10.00/2026-10-14 和 j2/1/25.00/2026-10-14。库存记录读取成功，但被类型条件排除。模式应显示整数 quantity、十进制价格和 date 类型的 order_date。
读取器对命名 JSON 字段使用声明的类型；与普通 CSV 记录不同，JSON 对象内的字段顺序不决定列身份。模式仍然定义预期约定。
从嵌套客户对象选择字段
在这一步中，你将读取额外的一层结构。查看客户事件文件：
cat /home/labex/project/data/customer-events.jsonl
customer 的值是另一个对象，包含 customer_id 和 region。在 Spark 模式中，它变成 struct（结构体），即包含命名子字段的字段。它不是数组，也不是需要关联的第二张表。
用内部 StructType 声明这个结构体。col("customer.region") 通过点分路径选择子字段，alias("region") 为所选结果设置简单列名。对客户标识符应用相同方法：
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/nested-json-orders.txt
cat /home/labex/project/output/nested-json-orders.txt
完整模式应显示 customer 是带两个字符串子字段的结构体。所选表应有 j3/c1/East/2/10.00/2026-10-15 和 j4/c2/West/1/25.00/2026-10-15。最终模式将 customer_id 和 region 呈现为普通字符串列，与数量、价格和日期并列。
你选择的值本来就在每个订单事件内部，无需对照表或关联。原始嵌套表示仍在 events 中，selected 则为后续操作提供更简单的平面视图。
对另一文件复用嵌套模式
在这一步中，你将改变交付数据，保持结构相同。先读取新文件：
cat /home/labex/project/data/new-customer-events.jsonl
预测输出：订单 j5，客户 c2，地区 West，数量 3，价格 12.50，日期 10 月 16 日。读取器和嵌套选择可以保持相同，只改变输入路径。运行完整版本：
cat > /home/labex/project/read_json_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("ReadJSONEvents").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
event_schema = StructType([
StructField("event_type", StringType(), True),
StructField("order_id", StringType(), True),
StructField("customer", StructType([
StructField("customer_id", StringType(), True),
StructField("region", StringType(), True),
]), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_date", DateType(), True),
])
events = (
spark.read.schema(event_schema)
.option("dateFormat", "yyyy-MM-dd")
.json("/home/labex/project/data/new-customer-events.jsonl")
)
events.printSchema()
orders = events.filter(col("event_type") == "order")
selected = orders.select(
"order_id",
col("customer.customer_id").alias("customer_id"),
col("customer.region").alias("region"),
"quantity", "unit_price", "order_date",
)
print("Selected customer orders:")
selected.orderBy("order_id").show(truncate=False)
selected.printSchema()
spark.stop()
PYTHON
bigdata-python /home/labex/project/read_json_orders.py > /home/labex/project/output/new-json-orders.txt
cat /home/labex/project/output/new-json-orders.txt
预期为预测中的单行结果，所选类型相同。约定结构未改变时，新文件无需重新设计模式。
三次读取都对已有文件使用 spark.read，并在脚本完成时结束，没有订阅未来消息。你已学会在不同时改变执行模型的情况下，改变输入格式。接下来，你将回到一批有意存在问题的订单数据，并定义明确清洗规则。
总结
你使用明确标量类型读取静态 JSON Lines，选择订单事件，并通过命名列路径展开简单客户结构体。第二批数据正确使用了相同模式。接下来，你将在清晰规则下处理缺失、重复和无效的订单记录。
