Introducción
Algunos sistemas entregan registros como filas CSV; otros entregan objetos JSON con campos con nombre. La empresa minorista proporciona ahora archivos de eventos con una etiqueta de tipo y valores relacionados con pedidos. Sus preguntas de procesamiento son conocidas: ¿qué significan los campos, qué registros pertenecen a pedidos y qué columnas debe exponer el resultado?
JSON Lines almacena un objeto JSON completo en cada línea. A diferencia de CSV, los nombres de campo acompañan a cada objeto y un objeto puede contener otro objeto. Primero leerá registros planos con tipos explícitos, después seleccionará campos de un objeto sencillo de cliente anidado y finalmente aplicará ese esquema a una segunda entrega.
Son archivos finitos leídos con spark.read, por lo que esto sigue siendo procesamiento por lotes. Que el nombre de un archivo contenga events no convierte una aplicación en procesamiento en flujo. El entorno y los datos de ejemplo son independientes; necesita los conceptos anteriores de esquema explícito, select, filter y alias, además de Python básico. No se requiere Kafka, Hadoop, Hive, configuración de flujos, arreglos ni recursión compleja. Utilice Terminal.
Lea JSON Lines plano y seleccione eventos de pedidos
En este paso, trasladará la lectura con esquema explícito de CSV a JSON. Examine el archivo proporcionado:
cd /home/labex/project
cat /home/labex/project/data/flat-events.jsonl
Cada línea es un objeto encerrado entre llaves. Los nombres entre comillas dobles identifican campos; dos puntos separan un nombre de su valor. Números como quantity son valores numéricos JSON, mientras que order_date es una cadena de fecha entre comillas que el DateType declarado interpretará. No hay línea de encabezado CSV ni un arreglo exterior.
El order_id, la cantidad, el precio y la fecha del primer objeto describen el mismo tipo de valores que un pedido CSV. El archivo también contiene un evento de inventario; leer un registro válido no significa que pertenezca a un resultado de pedidos. El esquema explícito siguiente declara los campos planos. json(path) lee este formato, mientras que el filtro conocido selecciona event_type == order.
Guarde el script completo con un documento en línea entre comillas y ejecútelo con el intérprete preparado de 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
La visualización inicial debería incluir los tres registros de eventos. Bajo Order events (eventos de pedidos), espere únicamente j1/2/10.00/2026-10-14 y j2/1/25.00/2026-10-14. El registro de inventario se leyó correctamente, pero la condición de tipo lo excluyó. El esquema debería mostrar cantidad integer, precio decimal y order_date de tipo date.
El lector utiliza los campos JSON con nombre con los tipos declarados; a diferencia de un registro CSV simple, el orden de campos dentro de cada objeto JSON no define la identidad de la columna. El esquema sigue definiendo el contrato previsto.
Seleccione campos de un objeto de cliente anidado
En este paso, leerá un nivel estructural adicional. Examine el archivo de eventos de clientes:
cat /home/labex/project/data/customer-events.jsonl
El valor customer es otro objeto que contiene customer_id y region. En el esquema de Spark esto se convierte en una estructura (struct), un campo que contiene campos hijos con nombre. No es un arreglo ni una segunda tabla que deba unirse.
Declare esa estructura con un StructType interior. col("customer.region") selecciona el campo hijo mediante su ruta con puntos; alias("region") da al resultado seleccionado un nombre sencillo de columna. Aplique la misma idea al identificador de cliente:
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
El esquema completo debería mostrar customer como una estructura con dos hijos de tipo cadena. La tabla seleccionada debería tener j3/c1/East/2/10.00/2026-10-15 y j4/c2/West/1/25.00/2026-10-15. Su esquema final expone customer_id y region como columnas ordinarias de cadenas junto con cantidad, precio y fecha.
Seleccionó valores que ya estaban dentro de cada evento de pedido. No se necesitó ninguna tabla de consulta ni unión. La representación anidada original permanece disponible en events, mientras que selected proporciona a las operaciones posteriores una vista plana más sencilla.
Reutilice el esquema anidado en otro archivo
En este paso, cambiará la entrega conservando la misma estructura. Lea primero el archivo nuevo:
cat /home/labex/project/data/new-customer-events.jsonl
Prediga la salida: un pedido j5 de c2 en West, cantidad 3 y precio 12.50 el 16 de octubre. El lector y la selección anidada pueden permanecer iguales; solo cambia la ruta de entrada. Ejecute esta versión completa:
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
Espere el resultado previsto de una fila y los mismos tipos seleccionados. Un archivo nuevo no exige inventar un esquema nuevo cuando la estructura acordada no cambia.
Las tres lecturas utilizaron spark.read sobre archivos existentes y terminaron cuando se completaron sus scripts. No se suscribieron a mensajes futuros. Ha aprendido a variar un formato de entrada sin cambiar simultáneamente el modelo de ejecución. A continuación, volverá a una entrega de pedidos deliberadamente imperfecta y definirá reglas explícitas de limpieza.
Resumen
Leyó JSON Lines estático con tipos escalares explícitos, seleccionó eventos de pedidos y aplanó una estructura sencilla de cliente mediante rutas de columnas con nombre. Una segunda entrega utilizó correctamente el mismo esquema. A continuación, tratará registros de pedidos ausentes, duplicados y no válidos siguiendo reglas claras de limpieza.
