Spark в кармане: рабочие промты, которые превращают PySpark-разработчика в команду дата-инженеров

Spark в кармане: рабочие промты, которые превращают PySpark-разработчика в команду дата-инженеров

PySpark-разработчик сегодня часто работает в одиночку: он и пишет трансформации, и ловит OutOfMemory, и разбирается с партиционированием Delta Lake, и считает стоимость кластера. LLM не заменяет инженера, но снимает рутину — если задавать ей правильные промты. Ниже — 10 конкретных промтов, которые можно копировать в чат с моделью (ChatGPT, Claude, Gemini) или в Databricks Assistant. Каждый промт — под свою задачу: отладка OOM, оптимизация джойнов, работа с Delta Lake, мониторинг и cost-оптимизация.

Как пользоваться подборкой

Промты написаны так, чтобы модель не гадала: в них есть роль, контекст (версия Spark, формат данных), ограничения и требуемый формат ответа. Подставляйте свои значения в квадратные скобки. Для Databricks Assistant промты работают так же — он построен на тех же принципах генерации кода и объяснений.

Промт 1. Диагностика OOM по логам

Для чего: когда джоб падает с java.lang.OutOfMemoryError или Container killed by YARN for exceeding memory limits.

Промт:

Ты — senior data engineer с опытом отладки Spark на YARN и Kubernetes.
Джоб на PySpark [версия, например 3.5] падает с ошибкой:
[вставь текст ошибки]
Конфигурация: executor memory [X], driver memory [Y], количество партиций [Z].
Трансформации: [кратко опиши: groupBy, join, window и т.д.].
Дай пошаговый план диагностики: какие метрики Spark UI смотреть (Shuffle Spill, GC time, peak execution memory), какие параметры менять и почему. Предложи 3 варианта фикса с плюсами и минусами.

Пример: джоб с groupBy("user_id").agg(sum("amount")) на 800 ГБ падал на executor с 8 ГБ. Модель указала на перекос ключей и предложила salting — добавление случайного суффикса к ключу. После правки джоб прошёл.

Промт 2. Поиск перекоса данных (data skew)

Для чего: когда часть тасков выполняется в разы дольше остальных.

Промт:

У меня PySpark-джоб с join по ключу [название]. В Spark UI один таск идёт [N] минут, остальные — секунды. Предложи способ найти перекошенные ключи: напиши код на PySpark, который считает топ-20 ключей по частоте и по объёму данных. Затем предложи два подхода: salted join и broadcast join, с условиями применимости каждого.

Пример: код вида df.groupBy(key).count().orderBy(desc("count")).show(20) быстро выявляет «тяжёлые» ключи. Broadcast подходит, если одна сторона меньше spark.sql.autoBroadcastJoinThreshold (по умолчанию 10 МБ).

Промт 3. Оптимизация join-стратегии

Для чего: когда джойн между большой и средней таблицей выполняется часами.

Промт:

Таблица A: [объём] ГБ, [строк] строк. Таблица B: [объём] ГБ, [строк] строк.
Джойн по [ключ], тип [inner/left]. Spark [версия].
Предложи оптимальную стратегию: broadcast, sort-merge или bucketed join. Объясни, как повлияет на shuffle. Напиши код с явным hint и покажи, как проверить план через explain("formatted").

Пример: для таблицы B в 50 МБ модель предложила broadcast(df_b) — это убрало shuffle и ускорило джойн. Проверка: df.explain("formatted") должен показать BroadcastHashJoin.

Промт 4. MERGE в Delta Lake

Для чего: upsert инкрементальных данных без дублей.

Промт:

Напиши PySpark-код для MERGE в Delta Lake: источник [описание], цель [описание], ключи [список].
Нужно: обновлять существующие строки, вставлять новые, не создавать дубли при повторном запуске (идемпотентность).
Покажи код с delta.tables.DeltaTable, условием merge и обработкой schema evolution. Укажи, какие настройки Delta нужны для mergeSchema.

Пример:

from delta.tables import DeltaTable
target = DeltaTable.forPath(spark, "/mnt/silver/users")
target.alias("t").merge(
    source.alias("s"), "t.user_id = s.user_id"
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

Промт 5. OPTIMIZE и Z-Order

Для чего: ускорить чтение Delta-таблицы, убрать мелкие файлы.

Промт:

Delta-таблица [путь], объём [X] ТБ, частые фильтры по колонкам [список].
Много мелких файлов из-за стриминга. Предложи план: OPTIMIZE, Z-ORDER BY, vacuum. Объясни, по каким колонкам Z-Order даст максимальный эффект и почему. Напиши команды SQL и PySpark.

Пример: OPTIMIZE events ZORDER BY (user_id, event_date) — типичный сценарий для таблиц событий. После Z-Order фильтры по этим колонкам читают меньше файлов.

Промт 6. Схема партиционирования

Для чего: выбрать партиции до того, как таблица разрослась.

Промт:

Данные: [объём], растут на [X] ГБ/день. Основные запросы фильтруют по [колонки], агрегируют по [колонки].
Предложи схему партиционирования для Delta Lake: по каким колонкам, с какой кардинальностью, нужен ли liquid clustering (Databricks). Объясни риск over-partitioning и дай правило выбора.

Пример: партиционирование по дате + liquid clustering по user_id в Databricks Runtime 13.3+ заменяет Z-Order и работает без перезаписи партиций.

Промт 7. Каркас идемпотентного пайплайна

Для чего: построить ETL-джоб, который можно безопасно перезапускать.

Промт:

Спроектируй PySpark-пайплайн: источник [описание], трансформации [список], приёмник [Delta].
Требования: идемпотентность, обработка late data, логирование, параметризация даты запуска. Напиши структуру кода с классами/функциями, конфигом и точкой входа. Используй Delta MERGE для идемпотентности.

Пример: структура с run_date как параметром и MERGE по ключу позволяет перезапускать джоб за любую дату без дублей.

Промт 8. Мониторинг джоба

Для чего: настроить алерты и метрики.

Промт:

PySpark-джоб в [Databricks/Airflow]. Предложи набор метрик для мониторинга: длительность, объём входных/выходных данных, число shuffle-операций, ошибки. Покажи, как логировать их через SparkListener или в Delta-таблицу. Дай пример кода с listener.

Пример: SparkListener собирает события onJobEnd, onStageCompleted — их можно писать в таблицу и строить дашборд.

Промт 9. Cost-оптимизация кластера

Для чего: снизить счёт за Databricks/EMR.

Промт:

Кластер: [тип инстансов], [N] воркеров, автскейл [вкл/выкл], runtime [версия]. Джобы: [описание].
Проанализируй, где переплата: idle-время, over-provisioning, неоптимальные инстансы. Предложи 5 конкретных изменений с ожидаемым эффектом. Учти spot-инстансы и photon.

Пример: переход на spot-инстансы для воркеров и включение Photon часто даёт заметную экономию без переписывания кода. Точные цифры зависят от нагрузки — считайте на своих метриках.

Промт 10. Объяснение чужого кода

Для чего: разобрать legacy-джоб без документации.

Промт:

Объясни, что делает этот PySpark-код, построчно. Укажи потенциальные проблемы: неоптимальные операции, неявные shuffle, ошибки с null. Предложи рефакторинг.
[вставь код]

Пример: модель часто находит collect() на больших данных, withColumn в цикле и count() для проверок — всё это дорогие операции.

Промт 11. Тесты для трансформаций

Для чего: покрыть логику тестами без кластера.

Промт:

Напиши pytest-тесты для PySpark-трансформаций [описание]. Используй локальную SparkSession и фикстуры. Покрой: пустой датафрейм, null-значения, дубли, граничные даты. Покажи пример с chispa или pandas-on-Spark.

Пример: chispa.assert_df_equality сравнивает датафреймы с учётом порядка строк.

Промт 12. Чтение streaming из Kafka

Для чего: построить near-real-time пайплайн.

Промт:

Напиши Structured Streaming-джоб: читает из Kafka топик [имя], парсит JSON-схему [описание], пишет в Delta с checkpoint. Нужна exactly-once семантика. Покажи код, настройки trigger и обработку watermark для late data.

Пример: writeStream.format("delta").option("checkpointLocation", ...) + trigger(processingTime="1 minute") — базовый каркас.

Промт 13. Миграция с Parquet на Delta

Для чего: перевести существующее хранилище.

Промт:

Есть таблицы в Parquet: [список], объёмы [X]. Нужно мигрировать в Delta Lake с сохранением истории.
Предложи пошаговый план: CONVERT TO DELTA, проверка, откат. Учти партиционирование и мелкие файлы. Напиши команды.

Пример: CONVERT TO DELTA parquet./path`` — конвертирует на месте без перезаписи данных.

Что дальше

Эти промты — не замена инженерному мышлению, а ускоритель: они снимают рутину диагностики и написания шаблонного кода, оставляя вам решения о стратегии. Начните с промта №1 и №10 — они чаще всего экономят часы. Дальше добавляйте свои контексты: версию Spark, размеры данных, типы кластеров — чем конкретнее вход, тем полезнее выход. Если хотите системно прокачать PySpark и Databricks — смотрите материалы asibiont.com/blog и практикуйтесь на реальных джобах.