{ "cells": [ { "cell_type": "markdown", "id": "471a32f6", "metadata": {}, "source": [ "# Модуль 3. Raw-данные и первое чтение в Spark\n", "\n", "В прошлом модуле мы проследили, как Spark работает с каталогом и файлами в MinIO. Теперь загрузим настоящие данные: поездки NYC Taxi и справочник зон.\n", "\n", "Сначала подготовим Spark-сессию и проверим подключение. Затем перенесем заранее скачанные файлы в raw-зону MinIO, прочитаем их через Spark и посмотрим на схему, пропуски и аномалии. Эти файлы останутся источником для таблиц, которые мы построим в следующих модулях.\n" ] }, { "cell_type": "markdown", "id": "723dba8f", "metadata": {}, "source": [ "## 0. Перед стартом\n", "\n", "- Подними стенд по `START_HERE.md` и пройди модули 1 и 2.\n", "- Скачай data bundle в `./data/nyc_taxi` по инструкции из `START_HERE.md`.\n", "- Открой этот ноутбук в новом ядре, чтобы начать без сессии от предыдущих опытов.\n" ] }, { "cell_type": "markdown", "id": "session-intro", "metadata": {}, "source": [ "## 1. Как настроен Spark в нашем стенде\n", "\n", "До сих пор мы пользовались готовым объектом `spark`. Разберем, как его создать и куда он отправляет задания: это понадобится при чтении файлов Taxi.\n", "\n", "`SparkSession` дает доступ к DataFrame и SQL. Его `SparkContext` связывает приложение с вычислительными ресурсами. В обычном ядре нашего ноутбука используется один работающий контекст.\n", "\n", "| Участник | Что делает в стенде |\n", "| --- | --- |\n", "| Driver | Работает в `jupyter`: выполняет Python-код, строит план и раздает задания |\n", "| Master | Принимает подключение по `spark://spark-master:7077` и выделяет ресурсы worker-ов |\n", "| Executors | Работают на `spark-worker-1` и `spark-worker-2`, читают данные и выполняют задания |\n", "\n", "В режиме `local[*]` задания выполняются в потоках на машине driver-а. Нам нужен кластер, поэтому проверим фактический master вместе с результатом вычисления.\n", "\n", "Создадим сессию. `.appName(...)` задает имя приложения в Spark UI, `.getOrCreate()` создает сессию или возвращает уже работающую. Адрес кластера пока возьмем из общего файла настроек.\n" ] }, { "cell_type": "code", "id": "session-demo", "metadata": {}, "source": [ "from pyspark.sql import SparkSession\n", "\n", "spark = (\n", " SparkSession.builder\n", " .appName(\"module-03-session-demo\")\n", " .getOrCreate()\n", ")\n", "spark.sparkContext.setLogLevel(\"ERROR\")\n", "\n", "print(\"Spark:\", spark.version)\n", "print(\"Master:\", spark.sparkContext.master)\n", "print(\"Приложение:\", spark.sparkContext.appName)\n", "assert spark.sparkContext.master == \"spark://spark-master:7077\", \"Нужен Spark-кластер стенда\"\n", "print(\"Сумма 0…9:\", spark.range(10).groupBy().sum(\"id\").first()[0])" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "session-config-explanation", "metadata": {}, "source": [ "### Откуда взялось подключение\n", "\n", "Master мы не указали в коде: Spark прочитал его из `spark/spark-defaults.conf`. Docker подключает этот файл к Jupyter как `/opt/spark/conf/spark-defaults.conf`. Там же подготовлены каталог Iceberg и доступ к MinIO.\n", "\n", "| Настройка в файле | Для чего нужна |\n", "| --- | --- |\n", "| `spark.sql.extensions` | Добавляет команды Iceberg к Spark SQL |\n", "| `spark.sql.catalog.lakehouse` | Подключает Spark-каталог с именем `lakehouse` |\n", "| `spark.sql.catalog.lakehouse.catalog-impl` и `.uri` | Выбирает JDBC-каталог в PostgreSQL, где зарегистрированы таблицы |\n", "| `spark.sql.catalog.lakehouse.warehouse` | Задает место для файлов таблиц: `s3a://lakehouse/warehouse` |\n", "| `spark.hadoop.fs.s3a.endpoint` и `.path.style.access` | Направляют S3A-запросы в MinIO внутри Docker-сети |\n", "\n", "При запуске через `spark-submit` настройки можно передать параметрами `--master`, `--name` и `--conf`. В коде для этого есть `.master(...)`, `.appName(...)` и `.config(...)`. Для нового приложения обычный приоритет: код -> параметры запуска -> файл defaults. Параметры запуска driver-а надо задать до его старта; уже работающий контекст новый builder не перестраивает.\n", "\n", "В нашем Jupyter дополнительные `--master` и `--conf` при запуске не передаются. Вне стенда потребуются также библиотеки Iceberg, S3A и JDBC: их состав и конфигурация описаны в [справочнике](../docs/stack_reference.md#spark).\n", "\n", "Посмотрим, какие значения получила сессия. В файле есть и учетные данные, поэтому выводим только выбранные несекретные настройки. Для SQL используем `spark.conf`; в Hadoop-настройках префикс `spark.hadoop.` уже снят.\n" ] }, { "cell_type": "code", "id": "session-config", "metadata": {}, "source": [ "for key in (\n", " \"spark.sql.extensions\",\n", " \"spark.sql.catalog.lakehouse\",\n", " \"spark.sql.catalog.lakehouse.catalog-impl\",\n", " \"spark.sql.catalog.lakehouse.uri\",\n", " \"spark.sql.catalog.lakehouse.warehouse\",\n", "):\n", " print(f\"{key} = {spark.conf.get(key)}\")\n", "\n", "hadoop_conf = spark.sparkContext._jsc.hadoopConfiguration()\n", "for key in (\"fs.s3a.endpoint\", \"fs.s3a.path.style.access\"):\n", " print(f\"{key} = {hadoop_conf.get(key)}\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "session-live-explanation", "metadata": {}, "source": [ "### Изменение настройки в работающей сессии\n", "\n", "Некоторые настройки можно менять, пока приложение работает. Например, `spark.sql.session.timeZone` определяет часовой пояс SQL-сессии. Посмотрим, как одна и та же секунда Unix-времени отображается в UTC и Москве.\n", "\n", "В Spark 3.5.1 эта настройка изменяется через `spark.conf.set(...)`. Сохраним исходное значение и вернем его в `finally`, в том числе при ошибке. Файлы данных этот опыт не меняет.\n" ] }, { "cell_type": "code", "id": "session-live-config", "metadata": {}, "source": [ "timezone_key = \"spark.sql.session.timeZone\"\n", "original_timezone = spark.conf.get(timezone_key)\n", "assert spark.conf.isModifiable(timezone_key)\n", "try:\n", " for timezone in (\"UTC\", \"Europe/Moscow\"):\n", " spark.conf.set(timezone_key, timezone)\n", " value = spark.sql(\"SELECT from_unixtime(0) AS displayed_time\").first()[0]\n", " print(timezone, value)\n", "finally:\n", " spark.conf.set(timezone_key, original_timezone)\n", "\n", "assert spark.conf.get(timezone_key) == original_timezone\n", "print(\"Исходный часовой пояс восстановлен:\", original_timezone)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "session-reuse-explanation", "metadata": {}, "source": [ "### Повторный вызов builder-а\n", "\n", "Теперь попробуем сменить имя приложения на `module-03-raw-ingest-and-first-read`, с которым будем читать Taxi. Проверим имя у работающего контекста: `spark.sparkContext.appName`.\n" ] }, { "cell_type": "code", "id": "session-reuse", "metadata": {}, "source": [ "demo_context = spark.sparkContext\n", "demo_name = demo_context.appName\n", "same_spark = (\n", " SparkSession.builder\n", " .appName(\"module-03-raw-ingest-and-first-read\")\n", " .getOrCreate()\n", ")\n", "assert same_spark is spark\n", "assert same_spark.sparkContext is demo_context\n", "assert same_spark.sparkContext.appName == demo_name\n", "print(\"То же приложение:\", same_spark.sparkContext.appName)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "session-stop-explanation", "metadata": {}, "source": [ "Имя осталось прежним: `getOrCreate()` вернул ту же сессию с тем же контекстом. Повтор ячейки сохраняет и переменные Python, и работающее приложение.\n", "\n", "Чтобы создать приложение с новым именем, вызовем `spark.stop()`. Этот вызов останавливает сессию и ее контекст. Переменные Python останутся, но старые Spark DataFrame использовать уже нельзя. Файлы MinIO и таблицы каталога сохранятся.\n", "\n", "Для смены master или настроек в общем файле перезапускай ядро: одной остановки сессии недостаточно для всех параметров запуска. Перезапуск удалит переменные Python, и тогда придется повторить импорты, создание сессии и чтение DataFrame.\n" ] }, { "cell_type": "code", "id": "session-stop", "metadata": {}, "source": [ "spark.stop()" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "session-exercise-explanation", "metadata": {}, "source": [ "## 2. Подключись к Spark самостоятельно\n", "\n", "Создай сессию, с которой продолжишь работу с Taxi. Заполни в следующей ячейке адрес кластера, имя `module-03-raw-ingest-and-first-read` и цепочку `.master(...)`, `.appName(...)`, `.getOrCreate()`.\n", "\n", "Настройки каталога возьми из общего файла. Адрес кластера в этот раз задай явно через `.master(...)`; остальные ноутбуки берут его из того же файла.\n", "\n", "Пока пропуски не заполнены, шаблон останавливается на `assert`. Выполни заполненный шаблон и проверку подключения дважды: имя и `applicationId` должны остаться теми же.\n" ] }, { "cell_type": "code", "id": "session-exercise", "metadata": {}, "source": [ "from pyspark.sql import SparkSession\n", "\n", "CLUSTER_MASTER = ...\n", "APP_NAME = ...\n", "assert CLUSTER_MASTER == \"spark://spark-master:7077\", \"Впиши адрес кластера\"\n", "assert APP_NAME == \"module-03-raw-ingest-and-first-read\", \"Впиши имя приложения\"\n", "\n", "# Замени ... цепочкой вызовов SparkSession.builder.\n", "spark = ...\n", "assert isinstance(spark, SparkSession), \"Создай SparkSession через builder\"\n", "spark.sparkContext.setLogLevel(\"ERROR\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "session-solution", "metadata": {}, "source": [ "
\n", "Контрольное решение (после своей попытки)\n", "\n", "```python\n", "from pyspark.sql import SparkSession\n", "\n", "CLUSTER_MASTER = \"spark://spark-master:7077\"\n", "APP_NAME = \"module-03-raw-ingest-and-first-read\"\n", "spark = (\n", " SparkSession.builder\n", " .master(CLUSTER_MASTER)\n", " .appName(APP_NAME)\n", " .getOrCreate()\n", ")\n", "spark.sparkContext.setLogLevel(\"ERROR\")\n", "```\n", "\n", "
" ] }, { "cell_type": "code", "id": "session-connection-check", "metadata": {}, "source": [ "assert spark.sparkContext.master == CLUSTER_MASTER\n", "assert spark.sparkContext.appName == APP_NAME\n", "print(\"Master:\", spark.sparkContext.master)\n", "print(\"Приложение:\", spark.sparkContext.appName)\n", "print(\"ID приложения:\", spark.sparkContext.applicationId)" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "session-table-explanation", "metadata": {}, "source": [ "### Запись и чтение через каталог\n", "\n", "Проверим, что сессия может работать с каталогом и MinIO. Следующая ячейка создаст собственную маленькую Iceberg-таблицу, прочитает из нее числа `0` и `1` и удалит ее в `finally`. Она работает и после удаления таблиц из модуля 2.\n", "\n", "У таблицы уникальное имя. `DROP TABLE ... PURGE` удаляет ее запись в каталоге, данные и метаданные; в MinIO могут остаться пустые маркеры `data/` и `metadata/`.\n", "\n", "Дождись завершения ячейки перед перезапуском ядра: при принудительной остановке процесса `finally` не выполнится. Если прервал ее, сохрани напечатанное имя, чтобы удалить эту таблицу после нового подключения.\n" ] }, { "cell_type": "code", "id": "session-table-check", "metadata": {}, "source": [ "from uuid import uuid4\n", "\n", "session_table = f\"lakehouse.default.session_practice_{uuid4().hex}\"\n", "print(\"Временная таблица:\", session_table)\n", "try:\n", " spark.range(2).writeTo(session_table).using(\"iceberg\").create()\n", " session_rows = [row.id for row in spark.table(session_table).orderBy(\"id\").collect()]\n", " assert session_rows == [0, 1]\n", " print(\"Прочитано через Iceberg:\", session_rows)\n", "finally:\n", " spark.sql(f\"DROP TABLE IF EXISTS {session_table} PURGE\")\n", "\n", "assert not spark.catalog.tableExists(session_table)\n", "print(\"Временная таблица удалена\")" ], "execution_count": null, "outputs": [] }, { "cell_type": "markdown", "id": "session-checkpoint", "metadata": {}, "source": [ "### Проверь запуск в новом ядре\n", "\n", "В выводе должны быть master `spark://spark-master:7077`, заданное имя приложения, строки `[0, 1]` и сообщение об удалении таблицы.\n", "\n", "Перезапусти ядро и повтори заполненный шаблон, проверку подключения и ячейку с таблицей. `applicationId` будет новым, остальные результаты должны совпасть. Импорт `SparkSession` уже есть в шаблоне, поэтому демонстрацию повторять не нужно.\n", "\n", "С этим подключением перейдем к файлам Taxi. Оставь сессию работать.\n" ] }, { "cell_type": "markdown", "id": "raw-preparation", "metadata": {}, "source": [ "## 3. Проверка data bundle\n", "\n", "Подготовим MinIO-клиент для загрузки файлов и пути к данным. Docker подключает каталог `data/` с хоста к Jupyter как `/opt/data/`.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "a5c56637", "metadata": {}, "outputs": [], "source": [ "from pathlib import Path\n", "import re\n", "from datetime import datetime\n", "\n", "import boto3\n", "import pandas as pd\n", "from botocore.exceptions import ClientError\n", "from pyspark.sql import SparkSession, functions as F\n", "\n", "DATA_DIR = Path(\"/opt/data/nyc_taxi\")\n", "BUCKET_NAME = \"lakehouse\"\n", "RAW_PREFIX = \"raw/nyc_taxi/\"\n", "RAW_URI = f\"s3a://{BUCKET_NAME}/{RAW_PREFIX}\"\n", "ZONE_LOOKUP_RAW_URI = f\"{RAW_URI}taxi_zone_lookup.csv\"\n", "\n", "s3 = boto3.client(\n", " \"s3\",\n", " endpoint_url=\"http://minio:9000\",\n", " aws_access_key_id=\"minioadmin\",\n", " aws_secret_access_key=\"minioadmin\",\n", ")\n", "\n", "\n", "def format_bytes(num_bytes: int) -> str:\n", " units = [\"B\", \"KB\", \"MB\", \"GB\", \"TB\"]\n", " value = float(num_bytes)\n", " for unit in units:\n", " if value < 1024 or unit == units[-1]:\n", " return f\"{value:.1f} {unit}\"\n", " value /= 1024\n", "\n", "\n", "def object_exists(bucket: str, key: str) -> bool:\n", " try:\n", " s3.head_object(Bucket=bucket, Key=key)\n", " return True\n", " except ClientError as exc:\n", " error_code = exc.response.get(\"Error\", {}).get(\"Code\", \"\")\n", " if error_code in {\"404\", \"NoSuchKey\", \"NotFound\"}:\n", " return False\n", " raise\n", "\n", "\n", "def list_objects(bucket: str, prefix: str) -> list[dict]:\n", " paginator = s3.get_paginator(\"list_objects_v2\")\n", " objects = []\n", " for page in paginator.paginate(Bucket=bucket, Prefix=prefix):\n", " objects.extend(page.get(\"Contents\", []))\n", " return objects\n", "\n", "\n", "def infer_month_bounds(paths: list[Path]) -> tuple[datetime, datetime]:\n", " months = []\n", " for path in paths:\n", " match = re.fullmatch(r\"yellow_tripdata_(\\d{4})-(\\d{2})\\.parquet\", path.name)\n", " if match:\n", " year = int(match.group(1))\n", " month = int(match.group(2))\n", " months.append((year, month))\n", "\n", " assert months, \"Не удалось определить месяцы из имён yellow_tripdata_*.parquet\"\n", "\n", " start_year, start_month = min(months)\n", " end_year, end_month = max(months)\n", "\n", " expected_start = datetime(start_year, start_month, 1)\n", " if end_month == 12:\n", " expected_end = datetime(end_year + 1, 1, 1)\n", " else:\n", " expected_end = datetime(end_year, end_month + 1, 1)\n", "\n", " return expected_start, expected_end\n", "\n", "\n", "print(\"Spark и MinIO-клиент готовы\")\n", "print(f\"RAW URI: {RAW_URI}\")" ] }, { "cell_type": "markdown", "id": "d4d8f302", "metadata": {}, "source": [ "Файлы должны быть скачаны заранее по `START_HERE.md`. Здесь проверим состав набора, прежде чем отправлять его в MinIO. Так во время урока нам не понадобится внешняя сеть, а при повторном прохождении мы будем читать те же исходные файлы.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "da2f7b51", "metadata": {}, "outputs": [], "source": [ "assert DATA_DIR.exists(), (\n", " \"Каталог /opt/data/nyc_taxi не найден. \"\n", " \"Сначала скачай data bundle по инструкции из START_HERE.md и убедись, что ./data смонтирован в Jupyter как /opt/data.\"\n", ")\n", "\n", "bundle_files = sorted(path for path in DATA_DIR.iterdir() if path.is_file())\n", "yellow_files = sorted(DATA_DIR.glob(\"yellow_tripdata_*.parquet\"))\n", "zone_lookup_file = DATA_DIR / \"taxi_zone_lookup.csv\"\n", "\n", "assert yellow_files, (\n", " \"Не найдены файлы yellow_tripdata_*.parquet в /opt/data/nyc_taxi. \"\n", " \"Скачай каноническое подмножество из START_HERE.md.\"\n", ")\n", "assert zone_lookup_file.exists(), (\n", " \"Не найден taxi_zone_lookup.csv. Он нужен и для этого модуля, и для самостоятельного задания. \"\n", " \"Скачай его по инструкции из START_HERE.md.\"\n", ")\n", "\n", "bundle_report = pd.DataFrame(\n", " [\n", " {\n", " \"file\": path.name,\n", " \"size_bytes\": path.stat().st_size,\n", " \"size_human\": format_bytes(path.stat().st_size),\n", " }\n", " for path in bundle_files\n", " ]\n", ").sort_values(\"file\").reset_index(drop=True)\n", "\n", "bundle_report\n" ] }, { "cell_type": "markdown", "id": "c8defd34", "metadata": {}, "source": [ "Набор файлов должен быть понятным и фиксированным. В этом модуле мы работаем не с «каким-то случайным parquet из интернета», а с заранее определённым учебным набором.\n" ] }, { "cell_type": "markdown", "id": "dabd90f2", "metadata": {}, "source": [ "## 4. Загрузка в raw-зону MinIO\n", "\n", "Raw-слой в Lakehouse похож на воспроизводимый входной `stg`-контур: мы сохраняем исходные файлы как есть, под оригинальными именами, не переписывая их содержимое.\n", "\n", "Конвенция пути в этом курсе:\n", "\n", "- bucket: `lakehouse`\n", "- raw prefix: `raw/nyc_taxi/`\n", "- полный путь для чтения: `s3a://lakehouse/raw/nyc_taxi/`\n", "\n", "Важно: raw-зона не смешивается с `warehouse/`, где потом будут жить Iceberg-таблицы.\n", "\n", "Повторный запуск этой секции должен быть безопасным:\n", "\n", "- если объекта ещё нет в raw-зоне, мы его загружаем;\n", "- если объект уже есть и размер совпадает, мы пропускаем загрузку;\n", "- если объект уже есть, но размер отличается, это сигнал проблемы с данными или bundle, и мы падаем сразу, без перезаписи raw.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "8644cc94", "metadata": {}, "outputs": [], "source": [ "upload_candidates = [*yellow_files, zone_lookup_file]\n", "upload_report = []\n", "\n", "for path in upload_candidates:\n", " object_key = f\"{RAW_PREFIX}{path.name}\"\n", " local_size = path.stat().st_size\n", " existed_before = object_exists(BUCKET_NAME, object_key)\n", "\n", " if existed_before:\n", " metadata = s3.head_object(Bucket=BUCKET_NAME, Key=object_key)\n", " remote_size = metadata[\"ContentLength\"]\n", " if remote_size != local_size:\n", " raise ValueError(\n", " \"Raw object already exists with different size: \"\n", " f\"{object_key} (local={local_size}, remote={remote_size}). \"\n", " \"Остановись и проверь data bundle вместо перезаписи raw-слоя.\"\n", " )\n", " action = \"skipped_same_size\"\n", " else:\n", " s3.upload_file(str(path), BUCKET_NAME, object_key)\n", " metadata = s3.head_object(Bucket=BUCKET_NAME, Key=object_key)\n", " remote_size = metadata[\"ContentLength\"]\n", " action = \"uploaded\"\n", "\n", " upload_report.append(\n", " {\n", " \"file\": path.name,\n", " \"raw_key\": object_key,\n", " \"action\": action,\n", " \"local_size_human\": format_bytes(local_size),\n", " \"remote_size_human\": format_bytes(remote_size),\n", " }\n", " )\n", "\n", "pd.DataFrame(upload_report)\n" ] }, { "cell_type": "code", "execution_count": null, "id": "11f4c459", "metadata": {}, "outputs": [], "source": [ "raw_objects = list_objects(BUCKET_NAME, RAW_PREFIX)\n", "\n", "pd.DataFrame(\n", " [\n", " {\n", " \"key\": obj[\"Key\"],\n", " \"size_human\": format_bytes(obj[\"Size\"]),\n", " }\n", " for obj in raw_objects\n", " ]\n", ")\n" ] }, { "cell_type": "markdown", "id": "48295c73", "metadata": {}, "source": [ "Если объект уже существовал, мы не перезаписываем его без необходимости. Совпадающий размер означает, что повторный запуск секции идемпотентен. Несовпадающий размер считается конфликтом и останавливает сценарий до любых изменений в raw.\n", "\n", "На этом шаге данные уже лежат в storage. `Spark` может читать их напрямую по `s3a://` пути, а для `Trino` понадобится зарегистрированная таблица. До неё мы дойдём в Модулях 4 и 6.\n" ] }, { "cell_type": "markdown", "id": "696d6adc", "metadata": {}, "source": [ "## 5. Первое чтение raw PARQUET через Spark\n", "\n", "Теперь читаем raw-файлы напрямую из `MinIO` по `s3a://` пути. Это и есть первая важная практика Lakehouse: вычислитель (`Spark`) работает поверх файлов в object storage, а не только поверх локальной файловой системы или таблиц внутри одной СУБД.\n", "\n", "Здесь важно различать два режима работы:\n", "\n", "- `spark.read.parquet(\"s3a://...\")` читает raw-файлы напрямую по физическому пути;\n", "- `spark.table(\"lakehouse....\")` читает уже зарегистрированную управляемую таблицу через каталог.\n", "\n", "Для проверки подключения мы читали маленькую таблицу по имени через `spark.table(...)`. Файлы Taxi пока не зарегистрированы в каталоге: прочитаем их по пути через `spark.read.parquet(...)`. Таблицу поездок создадим в модуле 4." ] }, { "cell_type": "code", "execution_count": null, "id": "0e8823aa", "metadata": {}, "outputs": [], "source": [ "single_file_uri = f\"{RAW_URI}yellow_tripdata_2024-01.parquet\"\n", "single_month_df = spark.read.parquet(single_file_uri)\n", "\n", "print(f\"Читаем один raw-файл: {single_file_uri}\")\n", "print(f\"Строк в одном файле: {single_month_df.count():,}\")\n" ] }, { "cell_type": "code", "execution_count": null, "id": "fd5ad623", "metadata": {}, "outputs": [], "source": [ "yellow_df = spark.read.parquet(f\"{RAW_URI}yellow_tripdata_*.parquet\")\n", "row_count = yellow_df.count()\n", "\n", "print(f\"Прочитано строк: {row_count:,}\")\n", "print(f\"Количество parquet-файлов в наборе: {len(yellow_files)}\")\n" ] }, { "cell_type": "code", "execution_count": null, "id": "5e147271", "metadata": {}, "outputs": [], "source": [ "yellow_df.show(10, truncate=False)\n" ] }, { "cell_type": "code", "execution_count": null, "id": "efa037e6", "metadata": {}, "outputs": [], "source": [ "input_files = sorted(yellow_df.inputFiles())\n", "pd.DataFrame({\"input_file\": input_files})\n" ] }, { "cell_type": "markdown", "id": "e8866f38", "metadata": {}, "source": [ "Обрати внимание: мы читаем сразу несколько файлов wildcard-путём. Это обычный рабочий сценарий для raw-слоя, где данные часто приходят партиями по датам, дням или месяцам.\n", "\n", "В этом модуле мы намеренно не используем `cache()` для всего raw DataFrame. Для учебного профилирования важнее устойчиво прочитать исходные файлы, чем пытаться держать весь raw-срез в памяти executor-ов.\n" ] }, { "cell_type": "markdown", "id": "9701728a", "metadata": {}, "source": [ "## 6. Схема, типы и null-значения\n", "\n", "Следующий шаг после первого чтения: понять, какие колонки пришли, какие у них типы и где уже на raw-слое встречаются `null`.\n", "\n", "Здесь важно не перепутать диагностику и очистку. В raw-слое мы наблюдаем и фиксируем реальность данных, а не исправляем её.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "e2809a8b", "metadata": {}, "outputs": [], "source": [ "yellow_df.printSchema()\n" ] }, { "cell_type": "code", "execution_count": null, "id": "126e5601", "metadata": {}, "outputs": [], "source": [ "schema_df = pd.DataFrame(yellow_df.dtypes, columns=[\"column\", \"spark_type\"])\n", "print(f\"Количество колонок: {len(yellow_df.columns)}\")\n", "schema_df\n" ] }, { "cell_type": "code", "execution_count": null, "id": "c6093e2a", "metadata": {}, "outputs": [], "source": [ "null_counts_row = yellow_df.select(\n", " *[F.sum(F.col(column).isNull().cast(\"int\")).alias(column) for column in yellow_df.columns]\n", ").toPandas().iloc[0]\n", "\n", "null_report = pd.DataFrame(\n", " {\n", " \"column\": yellow_df.columns,\n", " \"null_count\": [int(null_counts_row[column]) for column in yellow_df.columns],\n", " }\n", ")\n", "null_report[\"null_share\"] = (null_report[\"null_count\"] / row_count).round(6)\n", "null_report.sort_values([\"null_count\", \"column\"], ascending=[False, True]).reset_index(drop=True)\n" ] }, { "cell_type": "markdown", "id": "567156e6", "metadata": {}, "source": [ "`Null` в raw-данных сам по себе не является багом загрузки. Это нормальная часть профиля источника. Позже, на `bronze` и `silver`, мы будем принимать явные решения: какие поля очищать, что отбрасывать, а что нормализовать.\n" ] }, { "cell_type": "markdown", "id": "69e6b77e", "metadata": {}, "source": [ "## 7. Базовые метрики и аномалии\n", "\n", "Теперь быстро профилируем данные и ищем несколько очевидных проблемных паттернов:\n", "\n", "- отрицательный `fare_amount`;\n", "- нулевая дистанция при ненулевой сумме;\n", "- слишком большой `total_amount`;\n", "- `pickup` вне ожидаемого диапазона дат.\n", "\n", "Диапазон дат будем выводить из имён файлов, а не хардкодить руками.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "3a4ed596", "metadata": {}, "outputs": [], "source": [ "numeric_columns = [\n", " column\n", " for column, spark_type in yellow_df.dtypes\n", " if spark_type.startswith((\"tinyint\", \"smallint\", \"int\", \"bigint\", \"float\", \"double\", \"decimal\"))\n", "]\n", "\n", "yellow_df.select(*numeric_columns).describe().toPandas()\n" ] }, { "cell_type": "code", "execution_count": null, "id": "cc029c98", "metadata": {}, "outputs": [], "source": [ "required_columns = {\n", " \"fare_amount\",\n", " \"trip_distance\",\n", " \"total_amount\",\n", " \"tpep_pickup_datetime\",\n", " \"PULocationID\",\n", " \"DOLocationID\",\n", "}\n", "missing_columns = sorted(required_columns.difference(yellow_df.columns))\n", "assert not missing_columns, (\n", " \"В raw-схеме не хватает ожидаемых колонок: \"\n", " f\"{missing_columns}. Возможно, источник поменял формат, и модуль требует обновления.\"\n", ")\n", "\n", "expected_start, expected_end = infer_month_bounds(yellow_files)\n", "\n", "anomaly_report = pd.DataFrame(\n", " [\n", " {\n", " \"check\": \"fare_amount < 0\",\n", " \"count\": yellow_df.filter(F.col(\"fare_amount\") < 0).count(),\n", " },\n", " {\n", " \"check\": \"trip_distance = 0 and total_amount != 0\",\n", " \"count\": yellow_df.filter(\n", " (F.col(\"trip_distance\") == 0) & (F.col(\"total_amount\") != 0)\n", " ).count(),\n", " },\n", " {\n", " \"check\": \"total_amount > 1000\",\n", " \"count\": yellow_df.filter(F.col(\"total_amount\") > 1000).count(),\n", " },\n", " {\n", " \"check\": f\"pickup outside [{expected_start.date()}, {expected_end.date()})\",\n", " \"count\": yellow_df.filter(\n", " (F.col(\"tpep_pickup_datetime\") < F.lit(expected_start))\n", " | (F.col(\"tpep_pickup_datetime\") >= F.lit(expected_end))\n", " ).count(),\n", " },\n", " ]\n", ")\n", "\n", "anomaly_report\n" ] }, { "cell_type": "code", "execution_count": null, "id": "0cc5aaf7", "metadata": {}, "outputs": [], "source": [ "suspicious_rows = yellow_df.filter(\n", " (F.col(\"fare_amount\") < 0)\n", " | ((F.col(\"trip_distance\") == 0) & (F.col(\"total_amount\") != 0))\n", " | (F.col(\"total_amount\") > 1000)\n", " | (F.col(\"tpep_pickup_datetime\") < F.lit(expected_start))\n", " | (F.col(\"tpep_pickup_datetime\") >= F.lit(expected_end))\n", ")\n", "\n", "suspicious_rows.select(\n", " \"tpep_pickup_datetime\",\n", " \"passenger_count\",\n", " \"trip_distance\",\n", " \"fare_amount\",\n", " \"total_amount\",\n", " \"PULocationID\",\n", " \"DOLocationID\",\n", ").show(20, truncate=False)\n" ] }, { "cell_type": "markdown", "id": "9ac27204", "metadata": {}, "source": [ "Реальные данные почти всегда содержат выбросы, странные записи и неоднозначные значения. Это не повод немедленно переписывать raw-файлы. Наоборот, raw нужен именно затем, чтобы исходный срез можно было перечитать и переработать заново с новыми правилами.\n" ] }, { "cell_type": "markdown", "id": "095c9ac6", "metadata": {}, "source": [ "## 8. Самостоятельное задание\n", "\n", "Используй raw-данные, которые уже лежат в `MinIO`:\n", "\n", "1. Прочитай `taxi_zone_lookup.csv` из raw-зоны через `Spark`.\n", "2. Выведи схему и первые 10 строк.\n", "3. Посчитай количество зон по `Borough`.\n", "4. Найди `top-5` `pickup`-локаций в поездках и присоедини к ним названия зон.\n", "5. Найди ещё одну аномалию или интересный паттерн и коротко опиши его.\n", "\n", "Подсказка: CSV-файл лежит по пути `s3a://lakehouse/raw/nyc_taxi/taxi_zone_lookup.csv`.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "fe530d7d", "metadata": {}, "outputs": [], "source": [ "# Ваш код: прочитай taxi_zone_lookup.csv из raw-зоны через Spark\n" ] }, { "cell_type": "code", "execution_count": null, "id": "99fb5256", "metadata": {}, "outputs": [], "source": [ "# Ваш код: выведи схему и первые 10 строк\n" ] }, { "cell_type": "code", "execution_count": null, "id": "219a40fc", "metadata": {}, "outputs": [], "source": [ "# Ваш код: посчитай количество зон по Borough и top-5 pickup locations с join на lookup\n" ] }, { "cell_type": "code", "execution_count": null, "id": "5c944f60", "metadata": {}, "outputs": [], "source": [ "# Ваш код: найди ещё одну аномалию или интересный паттерн в данных\n" ] }, { "cell_type": "markdown", "id": "e23dfe47", "metadata": {}, "source": [ "## 9. Почему raw-слой важен\n", "\n", "Raw-слой полезен не потому, что в нём удобно жить постоянно, а потому что он даёт устойчивую исходную точку. Если завтра поменяются правила очистки, дедупликации или нормализации, мы не обязаны заново скачивать источник и гадать, что именно уже успели «подправить». Мы просто перечитываем raw и перестраиваем downstream-слои.\n", "\n", "Для инженера с опытом `PostgreSQL / Greenplum` полезна такая аналогия: raw похож на аккуратный внешний `staging`-контур, но в Lakehouse он обычно хранится в object storage как набор файлов, а не как таблица внутри одной СУБД.\n", "\n", "Отдельно важно, что загрузка источника вынесена из ноутбука в onboarding. Это делает практику менее хрупкой: ноутбук не зависит от внешней сети, а набор данных остаётся одинаковым у всех студентов.\n", "\n", "Наконец, raw не стоит чинить «на месте». Если в исходных данных есть `null`, отрицательные значения или странные даты, это сигнал для явных трансформаций в `bronze` и `silver`, а не повод тихо переписать входные файлы.\n" ] }, { "cell_type": "markdown", "id": "8849c73f", "metadata": {}, "source": [ "## 10. Checkpoint\n", "\n", "Проверь себя:\n", "\n", "1. К какому master подключена сессия? Какие настройки ты задал в коде, а какие она получила из общего файла?\n", "2. Чем SparkSession отличается от SparkContext? Где работают driver и executors?\n", "3. Почему часовой пояс изменился сразу, а для нового имени приложения понадобилось остановить сессию? Что придется повторить после перезапуска ядра?\n", "4. Можешь ли ты показать загруженные raw-файлы в MinIO и прочитать их через Spark?\n", "5. Сколько колонок в Yellow Taxi, какие у них типы и где встречаются `null`? Почему пропуски не обязательно означают ошибку загрузки?\n", "6. Зачем сохранять raw-слой без изменений, если дальше будут трансформации?\n", "7. Почему мы скачали датасет до запуска ноутбука? Что потребуется проверить, если источник изменит формат файлов?\n" ] }, { "cell_type": "markdown", "id": "371c4c9d", "metadata": {}, "source": [ "## 11. Завершение\n", "\n", "Мы намеренно не удаляем raw-данные из `MinIO`. В Модуле 4 они понадобятся для построения первого управляемого `bronze`-слоя на `Iceberg`.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "8574e96b", "metadata": {}, "outputs": [], "source": [ "spark.stop()\n" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.10.12" } }, "nbformat": 4, "nbformat_minor": 5 }