968 lines
44 KiB
Plaintext
968 lines
44 KiB
Plaintext
{
|
||
"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": [
|
||
"<details>\n",
|
||
"<summary>Контрольное решение (после своей попытки)</summary>\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",
|
||
"</details>"
|
||
]
|
||
},
|
||
{
|
||
"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
|
||
}
|