Introducción
La empresa minorista ya tiene un conjunto correcto y con tipos de resultados de pedidos pagados. Ahora otra aplicación necesita leer esos resultados más adelante, seleccionar una fecha y confiar en que una exportación repetida no haya duplicado pedidos silenciosamente. Guardar datos forma parte del contrato de procesamiento, no es simplemente la última línea de un script.
Escribirá Parquet con particionamiento de directorios por order_date. La raíz del conjunto contendrá subdirectorios con nombres de fecha y Spark reconstruirá la columna de fecha al leer esa raíz. Estos directorios de almacenamiento son diferentes de las particiones de cálculo utilizadas para ejecutar trabajo; el siguiente laboratorio examina la ejecución por separado.
Este laboratorio prepara de forma independiente cuatro pedidos pagados con importes decimales en una entrada local Parquet. Debe conocer lecturas de DataFrame, filtros y el flujo mínimo de sobrescritura Parquet. No se requiere catálogo de Hive, HDFS, conocimientos previos de SQL, instalación ni salidas anteriores. Utilice Terminal.
Escriba un conjunto particionado por fecha y vuelva a leerlo
En este paso, organizará los resultados preparados en directorios de fecha. El conjunto de entrada de data/prepared-orders contiene cuatro pedidos, con los mismos tipos de cantidad, precio decimal, fecha e importe decimal order_amount utilizados anteriormente. Parquet es binario; examínelo mediante Spark, en lugar de esperar que cat muestre texto CSV.
Comience en su espacio de trabajo:
cd /home/labex/project
partitionBy("order_date") indica al escritor que coloque las filas en directorios nombrados por fecha. mode("overwrite") sustituye el conjunto derivado existente en la ruta de salida especificada. Mantenga esa ruta de salida separada de input_path, que debe permanecer intacta.
Guarde y ejecute el programa completo:
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
Espere cuatro registros reconstruidos: o1 y o2 el 14 de octubre, y o4 y o6 el 15 de octubre, con importes 20.00, 25.00, 50.00 y 30.00. Main rows debería ser 4. Compruebe que quantity sea integer, el precio y el importe sean decimales y order_date sea date.
Una columna de partición puede aparecer en otra posición después de la lectura porque Spark la reconstruye a partir de los nombres de directorio en lugar de almacenarla como un campo ordinario en cada archivo de partición. Los nombres y valores de columnas establecen su identidad; utilice un select explícito cuando un consumidor posterior requiera un orden concreto de columnas. El orden de presentación también necesita orderBy.
Examine directorios y seleccione una fecha
En este paso, relacionará la estructura de directorios con la columna de fecha reconstruida. Liste la salida de forma recursiva; -R incluye los subdirectorios de fecha y sus archivos:
ls -R /home/labex/project/output/orders-by-date
Espere los directorios order_date=2026-10-14 y order_date=2026-10-15. Los nombres de archivos part, su número y los metadatos ocultos pueden variar; no fije un archivo part concreto como si fuera todo el conjunto. La raíz significativa es orders-by-date.
Leer esa raíz permite a Spark descubrir los valores de partición. Este descubrimiento de la estructura de archivos no requiere registrar una partición ni tabla de Hive. El siguiente script independiente lee la raíz, compara su columna DateType con una cadena válida de fecha ISO y guarda por separado los registros del día seleccionado:
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
Espere únicamente o1 y o2, con sus fechas del 14 de octubre e importes 20.00 y 25.00. La otra fecha sigue existiendo en el conjunto principal. Una lectura filtrada cambia los registros devueltos; no elimina los directorios no seleccionados.
Examinó la organización del almacenamiento y el resultado real de consulta. En este ejemplo diminuto, el tiempo transcurrido no demuestra que un formato o una estructura sean más rápidos. El propósito es un conjunto correcto y descubrible que puedan utilizar lectores posteriores.
Compare la sobrescritura repetida con el anexado de forma segura
En este paso, conservará el conjunto principal correcto y experimentará en un directorio separado. Prediga los conteos desechables: escribir las mismas cuatro filas con sobrescritura dos veces deja cuatro filas cada vez; añadir después una copia completa produce ocho filas. Esas filas adicionales son exportaciones duplicadas, no pedidos nuevos.
Ejecute esta versión completa. La ruta principal utiliza sobrescritura. Solo write-mode-demo recibe el anexado deliberado:
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
Busque Main rows: 4, First overwrite rows: 4, Second overwrite rows: 4 y After append rows: 8. El conjunto desechable contiene ahora intencionalmente dos copias de cada pedido y muestra por qué añadir una instantánea completa es inadecuado para la exportación principal.
Vuelva a ejecutar el programa guardado para comprobar la repetibilidad del flujo completo:
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
Deberían aparecer los mismos conteos. Cada ejecución completa reconstruye el conjunto principal y reinicia el experimento desechable antes del anexado intencional, por lo que no crece a doce ni dieciséis filas. Añadir es una política distinta para datos realmente nuevos y sin solapamiento; la sobrescritura es la política elegida para esta instantánea derivada completa.
El conjunto original prepared-orders sigue disponible. Los directorios de fecha describen valores almacenados, mientras que las particiones de cálculo de Spark en ejecución describen cómo divide el trabajo un trabajo de Spark. A continuación, relacionará ese segundo significado de partición con los planes y la ejecución real.
Resumen
Escribió un conjunto Parquet particionado por fecha, reconstruyó sus registros con tipos desde la raíz y seleccionó una fecha sin eliminar las demás. La sobrescritura repetida conservó la instantánea principal, mientras que un anexado desechable demostró copias duplicadas. A continuación, los planes de ejecución y los eventos reales explicarán las particiones de cálculo, los trabajos, las etapas y las tareas.
