Introducción
Ha escrito transformaciones de Spark, activado acciones y guardado datos. Ahora relacionará esas operaciones conocidas con evidencia de ejecución: un plan que describe el trabajo y un registro real de eventos que documenta trabajos, etapas y tareas. El objetivo es explicar el comportamiento y leer errores útiles, en lugar de memorizar conteos internos de tareas.
El controlador (driver) ejecuta su aplicación y coordina el trabajo. En un clúster, los ejecutores ejecutan tareas con recursos asignados; local[2] ejecuta el trabajo de Spark en esta única máquina con dos hilos de ejecución. Un trabajo puede dividirse en etapas y cada etapa ejecuta tareas sobre particiones de cálculo. Una acción puede causar más de un trabajo, especialmente con la planificación de consultas y el comportamiento adaptativo; los conteos exactos pueden variar.
Este laboratorio independiente proporciona pedidos limpios, clientes, dos expresiones deliberadamente incorrectas y una herramienta de solo lectura para resumir eventos. Debe conocer filtrado, agrupación, uniones y salidas Parquet. No ajustará recursos, no recopilará el conjunto general de datos en bucles Python ni desplegará un clúster. Utilice Terminal.
Relacione una consulta conocida con su plan de ejecución
En este paso, examinará una consulta diaria regional de ventas pagadas antes de solicitar sus resultados. El lector, el filtro, el importe, la unión de clientes y la agregación repiten habilidades anteriores.
Aquí groupBy("order_date", "region") utiliza dos claves juntas: cada combinación distinta de fecha y región forma un grupo. East el 14 de octubre está separado de East el 15 de octubre y de West el 14 de octubre. Los conteos y las sumas se calculan dentro de cada combinación. Esto combina las preguntas diarias y regionales anteriores en un informe. explain(mode="formatted") imprime los operadores planificados en lugar de devolver filas del informe.
La SparkSession también habilita un registro real de eventos en output/spark-events. La compresión está desactivada para que ese registro didáctico sea legible después de detenerse la aplicación. Estos ajustes registran la ejecución; no sustituyen registros de trabajos inventados. El directorio de destino ya está preparado.
Comience y ejecute el script completo de inspección:
cd /home/labex/project
cat > /home/labex/project/inspect_execution.py <<'PYTHON'
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, sum as spark_sum, spark_partition_id
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DecimalType, DateType
spark = (
SparkSession.builder.appName("InspectRetailExecution").master("local[2]")
.config("spark.eventLog.enabled", "true")
.config("spark.eventLog.dir", "file:///home/labex/project/output/spark-events")
.config("spark.eventLog.compress", "false")
.config("spark.eventLog.rolling.enabled", "false")
.getOrCreate()
)
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
customers = spark.read.option("header", True).schema("customer_id string, region string").csv("/home/labex/project/data/customers.csv")
paid = orders.filter(col("order_status") == "paid").withColumn("order_amount", col("quantity") * col("unit_price"))
matched = paid.alias("o").join(customers.alias("c"), col("o.customer_id") == col("c.customer_id"), "inner").select(
col("o.order_date").alias("order_date"), col("c.region").alias("region"), col("o.order_amount").alias("order_amount")
)
report = matched.groupBy("order_date", "region").agg(
count("*").alias("paid_order_count"), spark_sum("order_amount").alias("revenue")
)
print("Report execution plan:")
report.explain(mode="formatted")
print("Report action result:")
report.orderBy("order_date", "region").show(truncate=False)
report.write.mode("overwrite").parquet("/home/labex/project/output/daily-regional-sales")
print("Computation partition IDs represented by source rows:")
orders.select(spark_partition_id().alias("compute_partition")).dropDuplicates().orderBy("compute_partition").show(truncate=False)
spark.stop()
PYTHON
bigdata-python /home/labex/project/inspect_execution.py > /home/labex/project/output/execution-plan-and-result.txt
cat /home/labex/project/output/execution-plan-and-result.txt
En Report execution plan, busque las funciones de filtrado, proyección, unión, agregación y cualquier Exchange que mueva registros entre particiones de cálculo. Los nombres y la estructura del plan pueden variar con la optimización. Un Exchange es un límite de ejecución, no un nuevo directorio de salida.
Bajo Report action result, espere 14 de octubre / East / 1 / 20.00, 14 de octubre / West / 1 / 25.00, 15 de octubre / East / 1 / 50.00 y 15 de octubre / West / 1 / 30.00. El total sigue siendo 125.00. show y el escritor solicitan la ejecución real del trabajo descrito; la planificación y la gestión de metadatos de entrada también pueden realizar trabajo preparatorio.
La visualización final utiliza spark_partition_id para mostrar los ID de particiones de cálculo representados por estos registros de origen. dropDuplicates lista cada ID observado una vez. Los ID y su número observado pueden variar y las particiones vacías pueden no aparecer. Estos grupos de ejecución son diferentes de los directorios con nombres de fecha escritos por partitionBy en el laboratorio anterior. Uno organiza el trabajo; el otro organiza los datos almacenados.
Lea eventos reales de trabajos, etapas y tareas
En este paso, examinará el registro de eventos producido por la aplicación que acaba de ejecutar. La herramienta proporcionada lee registros JSON de eventos de ese directorio y resume eventos reales. No necesita escribir un analizador de registros ni seleccionar un nombre de archivo específico de una versión.
Ejecute la herramienta de resumen proporcionando el directorio de eventos como argumento de entrada. El > del shell conserva su salida para revisarla:
bigdata-python /home/labex/project/tools/summarize-events.py /home/labex/project/output/spark-events > /home/labex/project/output/event-summary.txt
cat /home/labex/project/output/event-summary.txt
Espere una identidad real de aplicación, conteos positivos de Jobs started, Stages submitted y Tasks ended y referencias de trabajos a etapas. Los números dependen de esta ejecución y pueden cambiar con la optimización, las divisiones de archivos y las versiones de Spark. Son evidencia observada, no objetivos que deba reproducir exactamente.
Una acción solicita una respuesta; Spark puede organizar el trabajo necesario en varios trabajos. Un trabajo hace referencia a etapas; las etapas contienen tareas para particiones de trabajo. Los límites de shuffle pueden separar etapas, mientras que varias transformaciones pueden combinarse dentro de una etapa. local[2] limita los hilos de ejecución simultáneos; no significa que cada aplicación tenga exactamente dos trabajos, etapas o tareas.
Con datos diminutos, el inicio de la JVM, la planificación y la gestión de archivos pueden resultar notorios respecto al cálculo en sí. Una pequeña diferencia de tiempo transcurrido no es una conclusión fiable de ajuste. Utilice primero el plan y la evidencia de eventos para comprender el trabajo.
Diagnostique y repare una columna desconocida
En este paso, leerá un error deliberado de resolución de nombres antes de corregir su causa. Examine el ejemplo incorrecto preparado:
cat /home/labex/project/tools/broken-column.py
El esquema declara quantity, mientras que su select final solicita quanity. Guarde la salida real del fallo. > redirige la salida normal al archivo de registro; 2>&1 envía el error estándar al mismo destino, por lo que también se guarda la traza. Se espera que el siguiente comando falle.
bigdata-python /home/labex/project/tools/broken-column.py > /home/labex/project/output/column-error.log 2>&1
tail -n 20 /home/labex/project/output/column-error.log
Busque el nombre de columna sin resolver y los nombres válidos sugeridos, si aparecen. Una traza larga contiene detalles de implementación, pero la referencia de columna señala la corrección. La entrada no necesita reinstalación ni conversión en un bucle Python.
Cree un programa corregido con el nombre de columna declarado, dejando disponible el ejemplo incorrecto de entrenamiento para compararlo:
cat > /home/labex/project/fixed_column.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("OrderExpressionExample").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
fixed = orders.select("order_id", "quantity")
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-quantities")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_column.py > /home/labex/project/output/fixed-column-result.txt
cat /home/labex/project/output/fixed-column-result.txt
Espere los seis ID con cantidades 2, 1, 1, 2, 1 y 2. La proyección ahora se resuelve contra el esquema real y guarda su resultado real. Los nombres de columnas, incluida su escritura exacta, forman parte del contrato de datos.
Repare un filtro que no es una condición booleana
En este paso, distinguirá una columna entera válida de una condición válida de filtrado. Examine el segundo ejemplo preparado:
cat /home/labex/project/tools/broken-filter.py
filter espera una expresión booleana para cada fila. col("quantity") por sí sola es un entero, no una prueba de verdadero o falso. Guarde el error deliberado de tipos incompatibles. Se espera que el siguiente comando falle.
bigdata-python /home/labex/project/tools/broken-filter.py > /home/labex/project/output/filter-error.log 2>&1
tail -n 20 /home/labex/project/output/filter-error.log
Busque la condición del filtro y su tipo entero no booleano. La reparación consiste en expresar la comparación prevista, quantity > 0, mediante el operador de columna ya aprendido. Ejecute el programa completo corregido:
cat > /home/labex/project/fixed_filter.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("OrderExpressionExample").master("local[2]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
order_schema = StructType([
StructField("order_id", StringType(), True),
StructField("customer_id", StringType(), True),
StructField("product_id", StringType(), True),
StructField("quantity", IntegerType(), True),
StructField("unit_price", DecimalType(10, 2), True),
StructField("order_status", StringType(), True),
StructField("order_date", DateType(), True),
])
orders = spark.read.option("header", True).schema(order_schema).csv("/home/labex/project/data/orders.csv")
fixed = orders.filter(col("quantity") > 0)
fixed.orderBy("order_id").show(truncate=False)
fixed.write.mode("overwrite").parquet("/home/labex/project/output/fixed-orders")
spark.stop()
PYTHON
bigdata-python /home/labex/project/fixed_filter.py > /home/labex/project/output/fixed-filter-result.txt
cat /home/labex/project/output/fixed-filter-result.txt
Las seis filas limpias del ejemplo tienen cantidades positivas, por lo que todas permanecen. Una cantidad negativa o cero no cumpliría esta condición; esta reparación concreta no aplica la regla separada de ventas pagadas.
Ahora tiene evidencia en tres niveles: la expresión y el esquema previstos, el trabajo realmente planificado y ejecutado y resultados persistentes correctos después de reparaciones específicas. Mantenga la distinción entre directorios de fecha almacenados y particiones de cálculo al explicar la ejecución. A continuación, enviará un script independiente por lotes con parámetros explícitos de entrada y salida.
Resumen
Relacionó transformaciones conocidas con un plan y un registro real de eventos, observó evidencia de aplicación, trabajos, etapas y tareas sin fijar sus conteos y distinguió particiones de cálculo de directorios de fecha. Reparó una columna sin resolver y un filtro no booleano mediante evidencia del esquema y los errores. A continuación, los parámetros explícitos de envío harán que su aplicación por lotes sea independiente de un proceso anterior.
