Введение
Вы написали преобразования Spark, вызвали действия и сохранили данные. Теперь вы свяжете знакомые операции со свидетельствами выполнения: планом, описывающим работу, и настоящим журналом событий заданий, этапов и задач. Цель — объяснять поведение и читать полезные ошибки, а не запоминать внутренние количества задач.
Драйвер запускает приложение и координирует работу. В кластере исполнители выполняют задачи на выделенных ресурсах; local[2] выполняет работу Spark на одной машине с двумя потоками. Задание (job) может делиться на этапы (stages), а каждый этап выполняет задачи (tasks) над вычислительными разделами. Одно действие может вызвать несколько заданий, особенно при планировании запросов и адаптивном поведении; точные количества различаются.
Независимая работа предоставляет чистые заказы, клиентов, два намеренно сломанных выражения и инструмент сводки событий только для чтения. Вам следует знать фильтрацию, группировку, соединения и результаты Parquet. Вы не будете настраивать ресурсы, собирать общий набор в циклы Python или развёртывать кластер. Используйте терминал.
Свяжите знакомый запрос с планом выполнения
На этом шаге вы изучите ежедневный региональный запрос оплаченных продаж до запроса результатов. Чтение, фильтр, сумма, соединение клиентов и агрегация повторяют прежние навыки.
Здесь groupBy("order_date", "region") использует два ключа вместе: каждая отдельная комбинация даты и региона образует группу. East за 14 октября отделён от East за 15 октября и от West за 14 октября. Количества и суммы рассчитываются внутри каждой комбинации. Это объединяет прежние ежедневный и региональный вопросы в один отчёт. explain(mode="formatted") выводит запланированные операторы вместо строк отчёта.
SparkSession также включает настоящий журнал событий в output/spark-events. Сжатие отключено, чтобы учебный журнал читался после остановки приложения. Настройки записывают выполнение, а не подставляют выдуманные записи заданий. Каталог назначения подготовлен для вас.
Начните и запустите полный скрипт изучения:
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
В Report execution plan найдите роли фильтрации, проекции, соединения, агрегации и Exchange, перемещающего записи между вычислительными разделами. Имена и структура плана могут различаться при оптимизации. Exchange — граница выполнения, а не новый выходной каталог.
Под Report action result ожидаются 14 октября / East / 1 / 20.00, 14 октября / West / 1 / 25.00, 15 октября / East / 1 / 50.00 и 15 октября / West / 1 / 30.00. Итог по-прежнему 125.00. show и запись запрашивают настоящее выполнение описанной работы; планирование и обработка входных метаданных тоже могут выполнять подготовительную работу.
Последний вывод использует spark_partition_id для показа ID вычислительных разделов, представленных исходными записями. dropDuplicates перечисляет каждый наблюдаемый ID один раз. ID и их число могут различаться, пустые разделы могут не появиться. Эти группы выполнения отличаются от каталогов дат, записанных partitionBy в предыдущей работе. Одно — организация работы, другое — структура сохранённых данных.
Прочитайте настоящие события заданий, этапов и задач
На этом шаге вы изучите журнал событий только что выполненного приложения. Предоставленный инструмент читает записи JSON из каталога и обобщает настоящие события. Писать разборщик журналов и выбирать имя файла, зависящее от версии, не нужно.
Запустите инструмент сводки, передав каталог событий входным аргументом. Символ оболочки > сохраняет вывод для проверки:
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
Ожидаются фактический идентификатор приложения, положительные количества Jobs started, Stages submitted и Tasks ended и ссылки заданий на этапы. Числа зависят от запуска и могут меняться с оптимизацией, разбиением файлов и версией Spark. Это наблюдения, а не цели для точного воспроизведения.
Действие запрашивает ответ; Spark может организовать нужную работу в несколько заданий. Задание ссылается на этапы; этапы содержат задачи для разделов работы. Границы shuffle могут разделять этапы, а несколько преобразований — объединяться внутри одного. local[2] ограничивает одновременные потоки выполнения; это не означает ровно два задания, этапа или задачи у каждого приложения.
На крошечных данных запуск JVM, планирование и работа с файлами могут быть заметны относительно самого вычисления. Небольшая разница времени не является надёжным выводом о настройке. Сначала используйте план и события для понимания работы.
Найдите и исправьте неизвестный столбец
На этом шаге вы прочитаете намеренную ошибку разрешения имени перед исправлением причины. Изучите подготовленный сломанный пример:
cat /home/labex/project/tools/broken-column.py
Схема объявляет quantity, а последний select запрашивает quanity. Сохраните фактический вывод сбоя. > перенаправляет обычный вывод в журнал; 2>&1 отправляет туда же стандартный поток ошибок, сохраняя и трассировку. Следующая команда должна завершиться ошибкой.
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
Найдите неразрешённое имя столбца и предлагаемые допустимые имена. Длинная трассировка содержит детали реализации, но ссылка на столбец указывает исправление. Входные данные не требуют переустановки или переноса в цикл Python.
Создайте исправленную программу с объявленным именем столбца, оставив сломанный учебный пример для сравнения:
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
Ожидаются все шесть ID с количествами 2, 1, 1, 2, 1 и 2. Теперь проекция разрешается по настоящей схеме и сохраняет фактический результат. Имена столбцов, включая написание, — часть контракта данных.
Исправьте фильтр без логического условия
На этом шаге вы отличите допустимый целочисленный столбец от допустимого условия фильтрации. Изучите второй подготовленный пример:
cat /home/labex/project/tools/broken-filter.py
filter ожидает логическое выражение для каждой строки. Один col("quantity") — целое число, а не проверка истины или лжи. Сохраните намеренное несоответствие типов. Следующая команда должна завершиться ошибкой.
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
Найдите условие фильтра и его нелогический целочисленный тип. Исправление — выразить нужное сравнение quantity > 0 уже изученным оператором столбца. Запустите исправленную полную программу:
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
Все шесть чистых строк примера имеют положительные количества, поэтому остаются. Отрицательное или нулевое количество не прошло бы условие; это конкретное исправление не применяет отдельное правило оплаченных продаж.
Теперь у вас свидетельства трёх уровней: нужное выражение и схема, фактически запланированная и выполненная работа и правильные сохранённые результаты после точечных исправлений. Объясняя выполнение, различайте каталоги дат и вычислительные разделы. Далее вы отправите независимый пакетный скрипт с явными параметрами входа и выхода.
Итоги
Вы связали знакомые преобразования с настоящим планом и журналом событий, изучили свидетельства приложения, заданий, этапов и задач без фиксации их числа и отличили вычислительные разделы от каталогов дат. Вы исправили неизвестный столбец и нелогический фильтр по схеме и ошибкам. Далее явные параметры отправки сделают пакетное приложение независимым от прежнего процесса.
