Introducción
Un script puede ejecutarse correctamente y aun así pasar registros incorrectos a un informe. La entrega de ventas de hoy contiene una fila de pedido repetida, un cliente ausente, texto numérico no válido, una cantidad no positiva, un precio negativo y una fecha mal formada. Rellenar automáticamente estos problemas con cero inventaría información comercial.
Separará tres decisiones: leer texto sin perder la evidencia original, interpretar de forma segura los campos numéricos y de fecha previstos y conservar únicamente las filas que cumplan una regla establecida. Después se eliminan los duplicados de filas válidas idénticas. Finalmente, guardará y volverá a leer un conjunto limpio de datos Parquet. Parquet es un formato binario columnar con tipos; este laboratorio enseña únicamente el flujo mínimo de escritura y lectura y reserva el almacenamiento particionado para más adelante.
El entorno preparado proporciona dos entregas independientes. Debe conocer la lectura CSV, las expresiones de columnas de Spark, el filtrado y los precios decimales. La expresión de conversión segura utilizada aquí es compatible con la versión preparada de Spark; no necesita cambiar su configuración de tratamiento de errores. Utilice Terminal. No se requieren archivos de salida anteriores, Hadoop, Hive ni instalación de servicios.
Conserve la entrega como datos intermedios de cadenas
En este paso, observará los defectos antes de convertir valores. Examine el origen local:
cd /home/labex/project
cat /home/labex/project/data/dirty-orders.csv
Hay catorce filas comerciales. Un DataFrame intermedio (staging) conserva estos campos de entrada como cadenas, lo que permite ver valores como two, unknown y not-a-date sin fingir que son números o fechas válidos. El tratamiento predeterminado del lector CSV convierte un campo de cliente vacío en NULL; NULL representa un valor ausente, no el texto "NULL".
Guarde y ejecute el programa intermedio completo. Como antes, el documento en línea PYTHON entre comillas escribe literalmente el código. show(30, truncate=False) permite mostrar hasta treinta registros y mantiene legible el texto de sus campos:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
staging.printSchema()
staging.orderBy("order_id").show(30, truncate=False)
print("Staging rows:", staging.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/staging-orders.txt
cat /home/labex/project/output/staging-orders.txt
Espere catorce registros mostrados y Staging rows: 14. Todas las columnas deberían ser cadenas. Busque la cantidad two de b2 y el precio unknown de b4; son evidencia útil, no errores ocultados por un valor predeterminado supuesto. o2 aparece dos veces y el estado de o1 tiene espacios alrededor.
El archivo original es de solo lectura en este ejercicio. Creará DataFrames derivados y directorios de salida dejando disponible la entrega original para su revisión.
Interprete valores de forma segura sin inventar valores predeterminados
En este paso, normalizará los dos campos de texto que necesitan retirar espacios exteriores e intentará convertir números y fechas. trim elimina espacios iniciales y finales. No adivina un cliente ausente ni cambia el estado de pago de un pedido.
expr permite proporcionar una expresión incorporada de Spark como texto. try_cast(quantity AS INT) intenta una conversión a número entero y produce NULL si ese valor no puede convertirse. Las expresiones equivalentes solicitan DECIMAL(10,2) para precios y DATE para fechas. Las conversiones ordinarias fallidas pueden generar errores bajo el comportamiento ANSI de Spark; esta conversión explícitamente segura evita depender de un ajuste global.
El cambio de esquema conserva las filas para que los problemas de interpretación sigan siendo observables. Ejecute esta versión:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
parsed.printSchema()
parsed.orderBy("order_id").show(30, truncate=False)
print("Parsed rows:", parsed.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/parsed-orders.txt
cat /home/labex/project/output/parsed-orders.txt
Espere Parsed rows: 14. La cantidad de b2, el precio de b4 y la fecha de b6 deberían ser ahora NULL. La cantidad cero de b3 y el precio negativo de b5 son valores numéricos sintácticamente válidos, pero siguen incumpliendo las reglas comerciales que aplicará a continuación. El estado de o1 debería ser ahora paid sin espacios alrededor.
Una conversión correcta no equivale a un pedido válido. Mantener separadas estas etapas permite explicar si una fila falló en la interpretación o en una restricción comercial. No se sustituyó ningún precio ausente por cero; el pedido real de precio cero o7 sigue siendo distinguible de un precio desconocido.
Aplique reglas explícitas de validez y duplicados
En este paso, conservará registros comerciales válidos y eliminará únicamente duplicados idénticos. Utilice estas reglas:
- customer_id está presente y no está vacío después de recortar espacios;
- quantity se interpretó correctamente y es mayor que cero;
- unit_price se interpretó correctamente y es al menos cero;
- order_date se interpretó correctamente;
- las filas idénticas en los siete campos normalizados se conservan una vez.
Los otros campos identificadores del ejemplo ya están bien formados. No tiene versiones contradictorias del mismo ID de pedido, por lo que este ejercicio no elige una versión ganadora entre estados cambiantes. Los registros pagados, pendientes y reembolsados permanecen si son válidos en los demás aspectos; limpiar una entrega es diferente de seleccionar ingresos pagados para un informe.
isNotNull comprueba explícitamente la presencia. Una comparación con NULL no se convierte en una coincidencia verdadera y filter conserva solo las condiciones verdaderas. Combinar las condiciones con & hace visible la política. dropDuplicates() sin un subconjunto elimina filas completas idénticas; utilizar únicamente order_id sería otra política que podría descartar información contradictoria.
Ejecute esta versión:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
staging = spark.read.option("header", True).csv("/home/labex/project/data/dirty-orders.csv")
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
print("Clean rows:")
clean.orderBy("order_id").show(30, truncate=False)
print("Staging count:", staging.count())
print("Valid count before deduplication:", valid.count())
print("Clean count:", clean.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/cleaning-result.txt
cat /home/labex/project/output/cleaning-result.txt
Espere Staging count: 14, Valid count before deduplication: 8 y Clean count: 7. Las reglas establecidas excluyen b1–b6 y se elimina una repetición de o2. Los siete ID conservados son o1 a o7.
El precio cero real del pedido o7 es válido bajo la regla de precio no negativo. o3 sigue pendiente y o5 reembolsado: ninguno de esos estados hace que el registro de origen esté mal formado. El conjunto resultante está limpio para esta política establecida, sin afirmar que se haya resuelto toda posible anomalía del mundo real.
Guarde, vuelva a leer y pruebe un defecto adicional
En este paso, hará reutilizable el conjunto limpio y probará la misma política en otra entrega. write.mode("overwrite").parquet(path) escribe un conjunto Parquet en un directorio y sustituye el contenido derivado anterior allí. Mantenga ese directorio separado de la entrada original. Spark puede crear varios archivos part y un marcador de finalización; no dependa de un nombre ni número concreto de archivos.
spark.read.parquet(path) lee el conjunto guardado y sus tipos almacenados. Consultar ese DataFrame leído de nuevo comprueba el archivo real, en lugar de limitarse a mostrar el plan anterior a la escritura. El programa completo siguiente coloca la lógica ya aprendida en una función de Python para que ambas entregas utilicen la misma regla de limpieza.
def clean_orders(input_path): define la función y nombra su parámetro de entrada. Las líneas con sangría son su cuerpo; se ejecutan cuando se llama a la función. La primera llamada proporciona el nombre de archivo original y la segunda proporciona el de la entrega adicional. Dentro de cada llamada, input_path hace referencia al valor proporcionado. return clean devuelve al llamador el DataFrame resultante; no guarda archivos. Las líneas de escritura sin sangría realizan esa acción por separado. Conserve la sangría al copiar el programa completo.
La entrega adicional contiene las catorce filas originales más una fila de cantidad negativa y otra copia exacta del pedido válido de precio cero. Prediga el resultado limpio: esas adiciones no deberían cambiar los siete registros conservados. Examine las adiciones:
tail -n 2 /home/labex/project/data/dirty-orders-extra.csv
Ejecute el programa completo de guardado y lectura:
cat > /home/labex/project/clean_orders.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, trim, expr
spark = SparkSession.builder.appName("CleanOrderDelivery").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
def clean_orders(input_path):
staging = spark.read.option("header", True).csv(input_path)
parsed = (
staging
.withColumn("customer_id", trim(col("customer_id")))
.withColumn("order_status", trim(col("order_status")))
.withColumn("quantity", expr("try_cast(quantity AS INT)"))
.withColumn("unit_price", expr("try_cast(unit_price AS DECIMAL(10,2))"))
.withColumn("order_date", expr("try_cast(order_date AS DATE)"))
)
valid = parsed.filter(
col("customer_id").isNotNull()
& (col("customer_id") != "")
& col("quantity").isNotNull() & (col("quantity") > 0)
& col("unit_price").isNotNull() & (col("unit_price") >= 0)
& col("order_date").isNotNull()
)
clean = valid.dropDuplicates()
return clean
clean = clean_orders("/home/labex/project/data/dirty-orders.csv")
clean.write.mode("overwrite").parquet("/home/labex/project/output/clean-orders")
reloaded = spark.read.parquet("/home/labex/project/output/clean-orders")
print("Read-back clean rows:")
reloaded.orderBy("order_id").show(30, truncate=False)
reloaded.printSchema()
print("Read-back count:", reloaded.count())
practice = clean_orders("/home/labex/project/data/dirty-orders-extra.csv")
practice.write.mode("overwrite").parquet("/home/labex/project/output/practice-clean-orders")
practice_reloaded = spark.read.parquet("/home/labex/project/output/practice-clean-orders")
print("Extra-delivery clean rows:")
practice_reloaded.orderBy("order_id").show(30, truncate=False)
print("Extra-delivery count:", practice_reloaded.count())
spark.stop()
PYTHON
bigdata-python /home/labex/project/clean_orders.py > /home/labex/project/output/clean-parquet-check.txt
cat /home/labex/project/output/clean-parquet-check.txt
Ambas tablas leídas deberían contener o1 a o7 exactamente una vez y ambos conteos deberían ser 7. El directorio principal de salida es output/clean-orders; practice-clean-orders es un experimento derivado separado. El esquema leído debería conservar cantidad integer, precio decimal y order_date de tipo date.
Este es un ciclo mínimo de escritura y lectura Parquet, no una comparación de velocidad ni un ejercicio de ajuste del almacenamiento. Ahora tiene un pequeño conjunto limpio que otro proceso de Spark puede leer y evidencia de que una fila no válida y un duplicado nuevos no cambian silenciosamente el resultado.
Resumen
Conservó el texto original, utilizó conversiones seguras para exponer valores no válidos, aplicó una política explícita de valores ausentes y rangos y eliminó filas duplicadas idénticas. Conservó precios cero válidos y registros no pagados, y después verificó la salida real de Parquet con tipos y una segunda entrega. El siguiente desafío le permitirá aplicar de forma independiente estas habilidades de limpieza y cálculo de importes.
