Введение
Магазин теперь отличает просмотренную пару сеанс/товар от повторных событий. Следующий вопрос: у каких товаров велика доля сеансов с просмотром без допустимого добавления и сколько наблюдений поддерживает сравнение?
Вы создадите полную таблицу со знаменателем просмотренных сеансов, числом завершённых и незавершённых и долей незавершённых. Сохраните товары без допустимых добавлений и все равные результаты. Это повод для дальнейшего наблюдения, не доказательство, что товар или страница вызвали прекращение посещения.
Независимая лабораторная предоставляет завершённый модуль путей из предыдущей работы и новые исходные файлы. Требуется знание её правил первого просмотра, пары сеанс/товар и времени, а также группировки Spark и Parquet. Используйте Terminal. Пакет конечен и закрыт; идентификаторы и секундные метки UTC корректны, повторные ID событий — точные копии. Авторская оценка — 35–45 минут. Потоковая обработка, оценка уникальных людей и причинные выводы не рассматриваются.
Постройте полное сравнение товаров
На этом шаге вы агрегируете одну строку на просмотренный путь в одну строку на товар. Начните в Terminal. Из предоставленных программ меняйте только analysis.py. Сохраните исходные файлы и предоставленные run_analysis.py, read_result.py и support.py без изменений; производные результаты размещайте в output/:
cd /home/labex/project
cat data/batch.csv
cat analysis.py
Модуль содержит read_events, build_views и build_journeys, независимо подготовленные из предыдущей лабораторной. Модуль хранит повторно используемые функции. Драйвер передаёт текущие SparkSession и путь ввода, затем возвращённый DataFrame путей в новую функцию build_report(journeys). Ничего из предыдущей VM или сеанса терминала не требуется.
Напомним контракт: полностью одинаковые копии событий удаляются; каждая просмотренная (session_id, product_id) использует первый просмотр; добавление допустимо только для обоих тех же ключей не раньше этого просмотра. Несколько допустимых добавлений всё равно дают один завершённый путь. Драйвер использует local[2], форматирование в UTC и один Python для драйвера и воркеров. Он пишет вывод после возврата функций.
Для товара с тремя просмотренными и двумя завершёнными парами посчитайте вручную: три сеанса с просмотром, два с корзиной, один без корзины, доля без корзины 1/3. Без этого знаменателя один визит мог бы выглядеть столь же убедительно, как большая выборка.
count("*") теперь считает строки путей, а не исходные события. Приведение Boolean completed к long даёт 1 для true и 0 для false. Spark sum суммирует их. Вычитание числа завершённых даёт число незавершённых; деление на положительное число просмотренных сеансов — долю от нуля до единицы. В каждой группе есть хотя бы один просмотр. Товар только с добавлениями не имеет группы или знаменателя; мы не выдумываем долю с нулевым знаменателем.
Напишите только агрегирование по товарам
Сохраните предоставленные read_events, build_views и build_journeys: это завершённая работа предыдущей лабораторной. Не набирайте их заново. F — их существующий псевдоним функций Spark. Откройте тот же модуль в nano:
nano analysis.py
Стрелками перейдите в конец build_journeys, затем оставьте пустую строку. Как уже изучено, Ctrl+W ищет имя функции; каждое новое объявление начинается у левого края, а тело имеет отступ четыре пробела. Вы напишете полный build_report, а не заполните подготовленный шаблон ответа.
Сначала сгруппируйте DataFrame путей по товару. В том же агрегировании получите два счётчика: все строки для знаменателя просмотров, затем сумму Boolean-признака завершения, приведённого к целому. Сохраните имена полей драйвера:
def build_report(journeys):
counts = journeys.groupBy("product_id").agg(
F.count("*").alias("view_sessions"),
F.sum(F.col("completed").cast("long")).alias("cart_sessions"),
)
Продолжайте внутри своей функции. withColumn выводит именованный столбец из сгруппированных счётчиков. Вычтите завершённые из просмотренных перед делением на тот же положительный знаменатель просмотров. Не фильтруйте сначала завершённые: это убрало бы товары без допустимых добавлений. Не выбирайте только один максимум: все товары с равной долей должны остаться в отчёте.
return counts.withColumn(
"no_cart_sessions", F.col("view_sessions") - F.col("cart_sessions")
).withColumn(
"no_cart_ratio", F.col("no_cart_sessions") / F.col("view_sessions")
)
Сохраните через Ctrl+O, подтвердите analysis.py клавишей Enter, затем выйдите через Ctrl+X. Снова откройте файл, чтобы изучить сохранённую функцию и убедиться, что предыдущие функции остались:
nano analysis.py
Через Ctrl+W найдите build_report. При ошибке стрелками и Backspace/Delete замените только затронутую строку; не добавляйте второе определение. Сохраните отступы продолжений и закрывающие скобки. Сохраните изменения, выйдите и прочитайте файл:
cat analysis.py
Если следующий запуск сообщит SyntaxError или IndentationError, найдите указанную строку в nano, проверьте окружающую функцию, скобки и отступы в четыре пробела, сохраните и повторите запуск. Успешно вычисленные, но неверные счётчик или доля требуют проверки группировки, приведения типа и знаменателя. Не сбрасывайте VM и не переписывайте рабочие функции G1 ради исправления отчёта.
Запустите стадию отчёта с явными путями ввода и вывода:
bigdata-python run_analysis.py data/batch.csv output/main --stage report
Ожидайте такую полную логическую таблицу; числа с плавающей точкой могут отображаться с большим количеством знаков:
product_id view_sessions cart_sessions no_cart_sessions no_cart_ratio
card 1 0 1 1.0
pin 1 0 1 1.0
tea 3 1 2 0.6666666666666666
book 2 1 1 0.5
mug 3 2 1 0.3333333333333333
Драйвер читает реально сохранённый Parquet перед отображением. Таблица сортируется по убыванию no_cart_ratio, затем по ID товара лишь для удобного чтения равенств. Остаются и card, и pin; равенство при сортировке не разрешает удалить один из них. Товар poster только с добавлением исключён, поскольку у него нет просмотренной пары.
Результаты — типизированные наборы в output/main/journeys и output/main/report. Последний содержит строковый ID товара, целочисленные счётчики и числовую долю. Названия и количество частей Parquet — детали реализации, не сущность отчёта. Эквивалентные выражения Spark DataFrame/SQL допустимы; сохраняйте конвейер от предоставленного ввода и выполняйте преобразования внутри Spark. Не собирайте исходные пути в Python, не переходите к RDD/pandas и не пересоздавайте их через checkpoint.
Прочитайте файлы и перестройте снимок
Прочитайте небольшой CSV и замените производный снимок, не меняя источник. Продолжайте в Terminal в /home/labex/project.
Драйвер сохраняет output/main/report.csv как один небольшой CSV UTF-8 для коллеги. Для сериализации он собирает лишь окончательную, уже агрегированную таблицу товаров после возврата build_report. Это финальное представление не заменяет преобразования Spark. Просмотрите настоящий файл:
cat output/main/report.csv
Заголовок содержит те же пять полей, что и Parquet-отчёт. Проверьте все пять товаров, нулевые числа сеансов с корзиной для card и pin и обе равные доли 1.0. CSV читается по строкам и полям; подсчёт строк или доверие имени файла не доказывают эти отношения.
Предоставленным считывателем прочитайте оба реальных набора Parquet и CSV. read_result.py читает указанный путь вывода, форматирует время внутри Spark в UTC и сравнивает CSV с сохранённым отчётом; он не создаёт и не дополняет отсутствующие результаты:
bigdata-python read_result.py output/main
Ожидайте десять строк путей, пять строк товаров и CSV matches stored report. Если файл отсутствует, повторите анализ с правильными вводом и назначением; считыватель намеренно завершается ошибкой, а не выдумывает ответ.
sha256sum печатает отпечаток содержимого каждого исходного файла. Шаблон *.csv выбирает CSV в data/. Сравните эти строки до и после перестроения:
sha256sum data/*.csv
bigdata-python run_analysis.py data/batch.csv output/main --stage report
bigdata-python read_result.py output/main
sha256sum data/*.csv
Тот же ввод даёт тот же логический отчёт без дублирования товаров или путей. Отпечатки исходников должны остаться прежними. Новые имена или байты файлов Parquet могут отличаться между сборками; контракт воспроизводимости относится к логическому содержимому. Запись заменяет только это известное производное назначение, не исходники и не другой каталог результатов.
Теперь используйте сохранённый модуль с обычным примером и отдельным назначением:
bigdata-python run_analysis.py data/normal.csv output/normal --stage report
bigdata-python read_result.py output/normal
Ожидайте две строки товаров, каждую с одним сеансом просмотра, одним с корзиной, нулём без корзины и долей 0. Основной результат остаётся целым. Одни параметризованные функции поддерживают оба пакета.
card и pin заслуживают внимания в этом маленьком пакете, но у каждого только один наблюдаемый сеанс. Высокая доля не объясняет прекращение визита, не определяет уникальных людей и не гарантирует сохранение закономерности. Счётчики и все равенства делают наблюдение проверяемым; причинное утверждение потребовало бы дополнительных данных.
Оценивание читает реальные рабочие результаты, не перестраивая их. Также оно загружает неизменённый модуль, вызывает функции на одноразовом пакете с другими ID и порядком строк, наблюдает реальное происхождение ввода Spark и читает временные результаты. Оно не оценивает точное написание команды, самостоятельно написанный план или маркер имени приложения и никогда не исправляет рабочий результат. Не добавляйте запись в функции; сериализацией управляет драйвер.
Резюме
Вы построили и прочитали полные сравнения Parquet и CSV, сохранили товары без добавлений и равенства и перестроили производный снимок, не изменив исходники. Знаменатель и счётчики объясняют каждую долю. Эти закономерности закрытых сеансов поддерживают дальнейшее наблюдение, не утверждение о причинности.
