Introduction
Some systems deliver records as CSV rows; others deliver JSON objects with named fields. The retailer now supplies event files containing a type label and order-related values. Your processing questions are familiar: what do the fields mean, which records belong to orders, and which columns should the result expose?
JSON Lines stores one complete JSON object on each line. Unlike CSV, field names travel with each object, and an object can contain another object. You will first read flat records with explicit types, then select fields from one simple nested customer object, and finally apply that schema to a second delivery.
These are finite files read with spark.read, so this is still batch processing. A filename containing events does not make an application streaming. The environment and fixtures are independent; you need the earlier explicit-schema, select, filter, and alias concepts plus basic Python. No Kafka, Hadoop, Hive, streaming setup, arrays, or complex recursion is required. Use the Terminal.
Read Flat JSON Lines and Select Order Events
In this step, you will transfer explicit-schema reading from CSV to JSON. Inspect the supplied file:
cd /home/labex/project
cat /home/labex/project/data/flat-events.jsonl
Each line is an object enclosed in braces. Double-quoted names identify fields; a colon separates a name from its value. Numbers such as quantity are numeric JSON values, while order_date is a quoted date string that the declared DateType will interpret. There is no CSV header line and no surrounding array.
The first object's order_id, quantity, price, and date describe the same kind of values as a CSV order. The file also contains an inventory event; reading a valid record does not mean it belongs in an order result. The explicit schema below declares the flat fields. json(path) reads this format, while the familiar filter selects event_type == order.
Save the complete script with a quoted here-document and run it through the prepared PySpark interpreter:
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
The initial display should include all three event records. Under Order events, expect only j1/2/10.00/2026-10-14 and j2/1/25.00/2026-10-14. The inventory record was read successfully but excluded by the type condition. The schema should show integer quantity, decimal price, and date order_date.
The reader uses the named JSON fields with the declared types; unlike a plain CSV record, field order inside each JSON object does not define the column's identity. The schema still defines the intended contract.
Select Fields from a Nested Customer Object
In this step, you will read one additional structural level. Inspect the customer-event file:
cat /home/labex/project/data/customer-events.jsonl
The customer value is another object, containing customer_id and region. In Spark's schema this becomes a struct, a field that contains named child fields. It is not an array or a second table to join.
Declare that struct with an inner StructType. col("customer.region") selects the child field by its dotted path; alias("region") gives the selected result a simple column name. Apply the same idea to the customer identifier:
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
The full schema should show customer as a struct with two string children. The selected table should have j3/c1/East/2/10.00/2026-10-15 and j4/c2/West/1/25.00/2026-10-15. Its final schema exposes customer_id and region as ordinary string columns alongside quantity, price, and date.
You selected values already inside each order event. No lookup table or join was required. The original nested representation remains available in events, while selected gives downstream operations a simpler flat view.
Reuse the Nested Schema on Another File
In this step, you will change the delivery while preserving the same structure. Read the new file first:
cat /home/labex/project/data/new-customer-events.jsonl
Predict the output: one order j5 for c2 in West, quantity 3 and price 12.50 on October 16. The reader and nested selection can stay the same; only the input path changes. Run this complete version:
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
Expect the predicted one-row result and the same selected types. A new file does not require inventing a new schema when the agreed structure is unchanged.
All three reads used spark.read over existing files and ended when their scripts completed. They did not subscribe to future messages. You have learned to vary an input format without simultaneously changing the execution model. Next, you will return to a deliberately imperfect order delivery and define explicit cleaning rules.
Summary
You read static JSON Lines with explicit scalar types, selected order events, and flattened a simple customer struct with named column paths. A second delivery used the same schema correctly. Next, you will handle missing, duplicate, and invalid order records under clear cleaning rules.
