Views и Temp Views: persistence, scoping и подводные камни

Три уровня изоляции (Temp / Global Temp / Persistent), раскрытие view в Catalyst, predicate pushdown, ловушка двойного вычисления, кэширование и security abstraction через Views.

core

Что такое View и зачем он нужен¶

View - это именованный SQL-запрос, сохранённый в каталоге. Никаких физических данных не создаётся: на диске хранится только текст запроса. Когда вы обращаетесь к view, Spark раскрывает его определение и выполняет исходный запрос.

Это похоже на функцию в программировании: вы один раз описываете логику и затем вызываете её по имени из любого места. Бизнес-правила, трансформации, фильтры - всё это инкапсулируется в view, а не дублируется в каждом SQL-запросе.

Зачем Views нужны в lakehouse¶

Без views аналитики работают напрямую с таблицами хранилища:

-- Без view: бизнес-логика дублируется в каждом запросе
SELECT
    customer_id,
    SUM(line_total) AS revenue,
    COUNT(*) AS orders
FROM silver.order_items
WHERE order_date >= DATE_SUB(CURRENT_DATE(), 90)
  AND status NOT IN ('cancelled', 'returned')
  AND is_test_order = FALSE
GROUP BY customer_id;

Эти условия - «последние 90 дней», «без отменённых», «без тестовых» - повторяются в десятках запросах. Если правило меняется (например, изменить 90 на 60 дней), нужно обновить каждый запрос. Это источник ошибок и расхождений между отчётами.

С view бизнес-логика определяется один раз:

-- Один раз: определяем "валидные заказы последнего квартала"
CREATE VIEW reporting.orders_last_quarter AS
SELECT *
FROM silver.order_items
WHERE order_date >= DATE_SUB(CURRENT_DATE(), 90)
  AND status NOT IN ('cancelled', 'returned')
  AND is_test_order = FALSE;

-- Любой аналитик: просто обращается к view
SELECT customer_id, SUM(line_total) AS revenue
FROM reporting.orders_last_quarter
GROUP BY customer_id;

Изменение правила: один ALTER VIEW или CREATE OR REPLACE VIEW - и все запросы автоматически используют новую логику.

View vs материализованное представление¶

Обычный Spark View не материализует данные. При каждом обращении к view выполняется исходный запрос заново. Это отличает его от Materialized View, который физически хранит результат запроса на диске и обновляется по расписанию.

В Spark нет нативных Materialized View (в отличие от Databricks, где они появились). Аналог - отдельная Gold-таблица, которую ETL-пайплайн перезаписывает по расписанию. View - всегда "живой" запрос к актуальным данным.

View vs DataFrame API¶

View и DataFrame - два способа описать одну трансформацию:

# DataFrame API: transformation только в Python-коде
orders_df = spark.table("silver.order_items") \
    .filter(col("status").isin("completed", "shipped")) \
    .filter(col("is_test_order") == False)

# View: transformation доступна как SQL-объект
spark.sql("""
    CREATE OR REPLACE TEMPORARY VIEW active_orders AS
    SELECT * FROM silver.order_items
    WHERE status IN ('completed', 'shipped')
      AND is_test_order = FALSE
""")

# Теперь можно из SQL:
spark.sql("SELECT * FROM active_orders WHERE order_date = '2024-01-15'")
# И из Python:
orders_df = spark.table("active_orders")

Выбор между ними зависит от контекста: если работаете в SQL-ориентированной среде (BI-инструменты, аналитики), view удобнее. Если вся логика в Python-коде одного пайплайна - DataFrame API достаточен.

Три уровня изоляции Views¶

Spark предоставляет три типа Views с разным временем жизни и областью видимости.

Local Temporary View - временная, только в этой сессии¶

Local Temp View привязана к конкретной SparkSession. Создаётся из DataFrame или через SQL:

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date

spark = SparkSession.builder.appName("views-demo").getOrCreate()

# Способ 1: из DataFrame
orders_df = spark.read.format("delta").load("s3a://datalake/silver/orders/")
orders_df.createOrReplaceTempView("orders")

# Способ 2: SQL
spark.sql("""
    CREATE OR REPLACE TEMPORARY VIEW active_orders AS
    SELECT *
    FROM orders
    WHERE status = 'completed'
      AND order_date >= '2024-01-01'
""")

# Использование
spark.sql("SELECT COUNT(*) FROM active_orders").show()

# Проверка существования
print(spark.catalog.tableExists("active_orders"))  # → True

# View умирает вместе с сессией
spark.stop()
# SparkSession закрыта → active_orders больше не существует

createOrReplaceTempView - идемпотентная операция. Если view с таким именем уже существует в этой сессии - она будет заменена новой без ошибки. createTempView (без Replace) - бросает ошибку при конфликте.

Важно: createOrReplaceTempView("name") без явной сессии регистрирует view только в той сессии, из которой вызван метод. В многопоточном коде, где каждый поток имеет свою SparkSession, view будет видна только в родительском потоке.

Global Temporary View - разделяется между сессиями¶

Global Temp View живёт в системной базе данных global_temp и видна из любой SparkSession в рамках одного SparkContext (одного JVM-процесса):

# Сессия 1
spark1 = SparkSession.builder.appName("session-1").getOrCreate()

expensive_df = spark1.sql("""
    SELECT user_id, SUM(amount) AS total_spent, COUNT(*) AS order_count
    FROM silver.orders
    GROUP BY user_id
""")

# Регистрируем как Global Temp View
expensive_df.createOrReplaceGlobalTempView("user_stats")

# Используем в сессии 1 - обязательно через префикс global_temp
spark1.sql("SELECT * FROM global_temp.user_stats LIMIT 5").show()

# Сессия 2 - другой поток, другая SparkSession
spark2 = SparkSession.builder.appName("session-2").getOrCreate()

# Видит global_temp.user_stats - та же JVM, тот же SparkContext
spark2.sql("SELECT * FROM global_temp.user_stats WHERE total_spent > 10000").show()

Типичные сценарии использования Global Temp View:

Jupyter Notebook с несколькими сессиями: общий справочник или промежуточный результат, на который ссылаются несколько ячеек с разными конфигурациями SparkSession
Многопоточный Airflow-таск: разные шаги одной задачи создают отдельные сессии, но должны видеть общие промежуточные данные
Стриминг + батч: streaming-сессия пишет агрегаты в Global Temp View, batch-сессия читает их для обогащения

Префикс global_temp. - обязательный. Если написать просто user_stats - Spark не найдёт view, потому что в базе данных по умолчанию её нет.

Persistent View - живёт в Metastore¶

Persistent View создаётся через CREATE VIEW и сохраняет определение в Hive Metastore (или другом постоянном каталоге). View переживает перезапуск приложения, смену сессий, перезапуск кластера - до тех пор, пока работает metastore.

-- В SQL
CREATE VIEW IF NOT EXISTS reporting.orders_last_quarter AS
SELECT
    o.order_id,
    o.customer_id,
    o.order_date,
    o.status,
    SUM(oi.line_total)       AS order_total,
    COUNT(oi.item_id)        AS item_count
FROM silver.orders o
JOIN silver.order_items oi USING (order_id)
WHERE o.order_date >= DATE_SUB(CURRENT_DATE(), 90)
  AND o.status NOT IN ('cancelled', 'returned')
GROUP BY o.order_id, o.customer_id, o.order_date, o.status;

# Или через PySpark
spark.sql("""
    CREATE OR REPLACE VIEW reporting.orders_last_quarter AS
    SELECT ...
""")

Требования к Persistent View:

Базовые таблицы должны быть зарегистрированы в том же каталоге (metastore)
Нельзя создать Persistent View поверх Temp View или локального DataFrame
Требуется enableHiveSupport() при создании SparkSession (или внешний catalog)

# РАБОТАЕТ: persistent view поверх persisted table
spark.sql("CREATE VIEW v AS SELECT * FROM silver.orders")

# НЕ РАБОТАЕТ: persistent view поверх temp view
spark.sql("CREATE TEMPORARY VIEW tmp AS ...")
spark.sql("CREATE VIEW pv AS SELECT * FROM tmp")
# → AnalysisException: Cannot create persistent view `pv` on top of temporary view `tmp`

Управление Persistent Views¶

-- Обновить определение
CREATE OR REPLACE VIEW reporting.orders_last_quarter AS ...;

-- Удалить
DROP VIEW IF EXISTS reporting.orders_last_quarter;

-- Просмотреть определение
SHOW CREATE TABLE reporting.orders_last_quarter;
-- Выведет: CREATE VIEW reporting.orders_last_quarter AS SELECT ...

-- Описание
DESCRIBE TABLE EXTENDED reporting.orders_last_quarter;
-- Type: VIEW, View Text: SELECT ...

Сравнительная таблица: три типа Views¶

Характеристика	Temp View	Global Temp View	Persistent View
Создание	`createOrReplaceTempView()`	`createOrReplaceGlobalTempView()`	`CREATE VIEW`
SQL	`CREATE TEMP VIEW`	`CREATE GLOBAL TEMP VIEW`	`CREATE VIEW`
Область видимости	Одна SparkSession	Все сессии одного SparkContext	Все приложения
Время жизни	Пока жива сессия	Пока жив SparkContext	До явного DROP
Metastore	Не требуется	Не требуется	Требуется
Доступ	`view_name`	`global_temp.view_name`	`db.view_name`
Может ссылаться на	Tables, Temp Views, DataFrames	Tables, Temp Views	Tables из metastore
Параллельный доступ	Нет	Да (в рамках JVM)	Да (любые приложения)

Под капотом: как Catalyst раскрывает Views¶

View Expansion в Logical Plan¶

Когда Spark встречает обращение к view, Catalyst подставляет тело view прямо в логический план запроса - это называется view expansion или view inlining. Никакого специального оператора "View" в физическом плане нет: после expansion план выглядит так, как будто вы написали запрос напрямую.

# Создадим view с нетривиальной логикой
spark.sql("""
    CREATE OR REPLACE TEMPORARY VIEW monthly_revenue AS
    SELECT
        DATE_TRUNC('month', order_date) AS month,
        category,
        SUM(line_total) AS revenue
    FROM silver.order_items
    WHERE status = 'completed'
    GROUP BY DATE_TRUNC('month', order_date), category
""")

# Запрос к view с внешней фильтрацией
result = spark.sql("""
    SELECT month, revenue
    FROM monthly_revenue
    WHERE category = 'electronics'
      AND month >= '2024-01-01'
""")

result.explain(mode="formatted")

Упрощённый план:

== Parsed Logical Plan ==
'Project ['month, 'revenue]
+- 'Filter ('category = 'electronics AND 'month >= '2024-01-01)
   +- 'UnresolvedRelation [monthly_revenue]   ← view по имени

== Analyzed Logical Plan ==
'Project ['month, 'revenue]
+- 'Filter ('category = 'electronics AND 'month >= '2024-01-01)
   +- 'Aggregate [DATE_TRUNC('month', order_date), category], ...  ← view раскрыт
      +- 'Filter (status = 'completed)
         +- 'UnresolvedRelation [silver, order_items]

== Optimized Logical Plan ==
Aggregate [month, category], [month, SUM(line_total) AS revenue]
+- Project [DATE_TRUNC('month', order_date) AS month, category, line_total]
   +- Filter (status = 'completed AND category = 'electronics  ← предикаты объединены
              AND DATE_TRUNC('month', order_date) >= 2024-01-01)
      +- Relation silver.order_items [...]

== Physical Plan ==
HashAggregate(keys=[month, category], ...)
+- Exchange hashpartitioning(month, category, 200)
   +- HashAggregate(keys=[month, category], ...)  ← partial
      +- Project [...]
         +- Filter (...)
            +- FileScan parquet silver.order_items [...]
               PushedFilters: [IsNotNull(status), EqualTo(status,completed),
                               IsNotNull(category), EqualTo(category,electronics)]

Ключевые наблюдения:

View раскрывается на этапе Parsed → Analyzed: UnresolvedRelation [monthly_revenue] заменяется полным деревом view-запроса
Предикаты опускаются (predicate pushdown): внешний WHERE category = 'electronics' объединяется с условием status = 'completed' из view и проталкивается к FileScan
В Physical Plan нет "View": выполняется чистый физический план без абстракций

Predicate Pushdown через Views¶

Catalyst умен: он протаскивает предикаты через view-абстракцию вниз, к физическим файлам. Фильтрация WHERE category = 'electronics' применяется при чтении Parquet, а не к результату view:

# Тест predicate pushdown через View
spark.sql("""
    CREATE OR REPLACE TEMPORARY VIEW all_orders AS
    SELECT order_id, order_date, category, line_total, status
    FROM silver.order_items
""")

# Внешний фильтр на view
plan = spark.sql("""
    SELECT SUM(line_total)
    FROM all_orders
    WHERE category = 'electronics'
      AND order_date = '2024-01-15'
""").explain(mode="extended")

В Physical Plan увидите:

FileScan parquet silver.order_items [...]
  PushedFilters: [EqualTo(category, electronics), EqualTo(order_date, 2024-01-15)]
  PartitionFilters: [isnotnull(order_date), EqualTo(order_date, 2024-01-15)]

Parquet-файлы читаются с предикатами - Spark даже не загружает другие партиции. View - прозрачная абстракция с точки зрения производительности предикатов.

Column Pruning через Views¶

Аналогично с projection (column pruning):

-- View с широкой схемой
CREATE TEMPORARY VIEW wide_orders AS
SELECT order_id, customer_id, order_date, status, category,
       line_total, quantity, unit_price, discount_pct,
       product_name, sku, _ingested_at, _source
FROM silver.order_items;

-- Запрос только двух колонок
SELECT order_id, line_total FROM wide_orders WHERE status = 'completed';

В Physical Plan:

FileScan parquet silver.order_items [order_id, status, line_total]
  ReadSchema: struct<order_id:bigint, status:string, line_total:decimal(14,2)>

Spark читает из Parquet только три нужных колонки, несмотря на то что view описывает 13. Column pruning работает сквозь view-абстракцию.

Вложенные Views: каскадное раскрытие¶

Каждый уровень вложенности добавляет один раунд expansion в Catalyst:

CREATE TEMP VIEW v1 AS
    SELECT * FROM silver.orders WHERE status = 'completed';

CREATE TEMP VIEW v2 AS
    SELECT customer_id, SUM(total) AS total FROM v1 GROUP BY customer_id;

CREATE TEMP VIEW v3 AS
    SELECT * FROM v2 WHERE total > 1000;

-- Запрос к v3 раскроет v3 → v2 → v1 → silver.orders
SELECT * FROM v3 WHERE customer_id < 50000;

Каскадное раскрытие происходит на этапе Analyzer. Catalyst в итоге видит один flat plan. Это нормально при 2–3 уровнях.

Проблема возникает при 5–7+ уровнях вложенности:

Время компиляции плана растёт экспоненциально: каждый уровень может добавить к плану тысячи узлов
Debugging становится невозможным: explain-план занимает сотни строк
Catalyst Rule Engine применяет оптимизационные правила ко всему развёрнутому плану - при сложном дереве это занимает секунды

Правило: не более 3 уровней вложенности views. При большей глубине - материализовать промежуточные результаты в Delta/Parquet.

Ловушка двойного вычисления¶

Это самый опасный подводный камень при работе с Views и Temp Views.

Суть проблемы¶

View - это только определение запроса. При каждом обращении к view Spark выполняет этот запрос заново. Если вы используете одну view дважды в одном запросе - запрос выполнится дважды.

# Создаём "тяжёлую" view с группировкой
spark.sql("""
    CREATE OR REPLACE TEMPORARY VIEW user_stats AS
    SELECT
        user_id,
        COUNT(*) AS total_events,
        SUM(amount) AS total_spent,
        MAX(event_ts) AS last_event
    FROM silver.events
    GROUP BY user_id
""")

# ОПАСНО: view используется ДВАЖДЫ
result = spark.sql("""
    SELECT
        a.user_id,
        a.total_spent,
        b.avg_spent
    FROM user_stats a
    JOIN (
        SELECT AVG(total_spent) AS avg_spent FROM user_stats  ← второе обращение
    ) b
    WHERE a.total_spent > b.avg_spent
""")

В DAG-графе Spark UI вы увидите: ветка вычисления user_stats появляется дважды - два отдельных HashAggregate узла, два FileScan. silver.events читается дважды, группировка выполняется дважды.

Для таблицы с миллиардом строк это означает: задача которая должна занять 5 минут займёт 10.

Диагностика через explain¶

result.explain(mode="formatted")

В выводе ищите одинаковые поддеревья:

== Physical Plan ==
SortMergeJoin [...]
:- HashAggregate(keys=[user_id], functions=[count, sum, max])  ← первое
:  +- Exchange hashpartitioning(user_id, ...)
:     +- HashAggregate(...)
:        +- FileScan parquet silver.events [...]   ← первый скан
+- HashAggregate(functions=[avg(total_spent)])     ← второе
   +- Exchange SinglePartition
      +- HashAggregate(keys=[user_id], ...)
         +- FileScan parquet silver.events [...]   ← второй скан (дубль!)

Два идентичных FileScan с двумя одинаковыми HashAggregate - явный признак double evaluation.

Решение 1: CTE вместо повторного обращения¶

Лучшее решение в большинстве случаев - переписать запрос с использованием CTE (WITH clause):

-- CTE: вычисляется один раз в рамках запроса
WITH user_stats AS (
    SELECT
        user_id,
        COUNT(*) AS total_events,
        SUM(amount) AS total_spent,
        MAX(event_ts) AS last_event
    FROM silver.events
    GROUP BY user_id
),
global_avg AS (
    SELECT AVG(total_spent) AS avg_spent FROM user_stats
)
SELECT
    s.user_id,
    s.total_spent,
    g.avg_spent
FROM user_stats s
CROSS JOIN global_avg g
WHERE s.total_spent > g.avg_spent;

CTE в Spark SQL - это не гарантия однократного вычисления. По умолчанию Catalyst может раскрыть CTE так же как View и вычислить его несколько раз. Но в большинстве версий Spark 3.x CTE с одним потребителем вычисляется один раз.

Гарантированное однократное вычисление - только материализация.

Решение 2: CACHE TABLE¶

# Материализуем результат view в памяти
spark.sql("CACHE TABLE user_stats")

# Теперь любые обращения к user_stats читают из кэша
result = spark.sql("""
    SELECT a.user_id, a.total_spent, b.avg_spent
    FROM user_stats a
    JOIN (SELECT AVG(total_spent) AS avg_spent FROM user_stats) b
    WHERE a.total_spent > b.avg_spent
""")

# После использования - освобождаем память
spark.sql("UNCACHE TABLE user_stats")

После CACHE TABLE в explain-плане появится оператор InMemoryRelation:

== Physical Plan ==
SortMergeJoin [...]
:- InMemoryTableScan [user_id, total_spent, ...]  ← из кэша
:  +- InMemoryRelation [...] (id=#1, ...)
+- InMemoryTableScan [avg_spent]                  ← тот же кэш
   +- InMemoryRelation [...] (id=#1, ...)         ← id совпадают!

Оба обращения к user_stats используют одну InMemoryRelation с одинаковым id - данные вычислены один раз.

Решение 3: DataFrame.persist()¶

Если работаете в Python API:

from pyspark.storagelevel import StorageLevel

user_stats_df = spark.sql("""
    SELECT user_id, COUNT(*) AS total_events, SUM(amount) AS total_spent
    FROM silver.events
    GROUP BY user_id
""")

# Материализуем в памяти (или memory+disk если не влезает)
user_stats_df.persist(StorageLevel.MEMORY_AND_DISK)
user_stats_df.count()  # trigger materialization

# Регистрируем как view для SQL-запросов
user_stats_df.createOrReplaceTempView("user_stats_cached")

# Оба обращения к user_stats_cached теперь читают из cache
result = spark.sql("""
    SELECT a.user_id, a.total_spent
    FROM user_stats_cached a
    JOIN (SELECT AVG(total_spent) AS avg_spent FROM user_stats_cached) b
    WHERE a.total_spent > b.avg_spent
""")

# Cleanup
user_stats_df.unpersist()

CACHE TABLE vs persist(): когда что¶

	CACHE TABLE	DataFrame.persist()
Контекст	SQL-первый пайплайн	Python-первый пайплайн
Trigger	Lazy (при первом запросе)	Eager с `.count()` / Lazy без
Видимость	SQL и Python	Только Python (если не создать view)
Освобождение	`UNCACHE TABLE`	`.unpersist()`
Storage level	Только MEMORY_AND_DISK	Любой StorageLevel

Schema Drift: как изменения таблицы ломают Views¶

Late Binding в Spark Views¶

В отличие от PostgreSQL или SQL Server, Spark использует late binding (позднее связывание) для Views. Схема view не проверяется при создании - только при выполнении. Это означает: view создастся успешно, даже если базовая таблица не существует (если TEMPORARY). Ошибка возникнет только при обращении к view.

Это создаёт риск: view существует в каталоге, но при запросе падает с ошибкой потому что базовая таблица изменилась.

Сценарий: удаление колонки ломает View¶

-- Шаг 1: View ссылается на конкретную колонку
CREATE VIEW reporting.customer_summary AS
SELECT
    customer_id,
    full_name,          -- ← эта колонка будет проблемой
    email,
    total_orders,
    total_spent
FROM silver.customers;

-- Шаг 2: через некоторое время команда решает переименовать колонку
-- в базовой таблице
ALTER TABLE silver.customers RENAME COLUMN full_name TO display_name;

-- Шаг 3: теперь View сломана
SELECT * FROM reporting.customer_summary;
-- AnalysisException: Resolved attribute(s) full_name#123 missing from
-- display_name#456, customer_id#457, ...

View хранит текст исходного запроса. После переименования колонки Catalyst пытается найти full_name в схеме таблицы, не находит - ошибка.

Как обнаружить сломанные Views¶

def validate_all_views(spark: SparkSession, database: str) -> list:
    """Находит все broken views в базе данных."""
    broken = []
    for table in spark.catalog.listTables(database):
        if table.tableType == "VIEW":
            try:
                spark.sql(f"SELECT 1 FROM {database}.{table.name} LIMIT 0")
            except Exception as e:
                broken.append({
                    "view": f"{database}.{table.name}",
                    "error": str(e)[:200]
                })
    return broken


# Проверяем все views в reporting
issues = validate_all_views(spark, "reporting")
for issue in issues:
    print(f"BROKEN: {issue['view']}")
    print(f"  Error: {issue['error']}")

Защита от Schema Drift: явные колонки вместо SELECT *¶

-- ПЛОХО: SELECT * внутри view
CREATE VIEW reporting.customers AS
SELECT * FROM silver.customers;
-- При добавлении новой колонки в silver.customers она автоматически
-- появляется в reporting.customers - может сломать downstream-запросы,
-- которые ожидают фиксированное число колонок

-- ХОРОШО: явный список колонок
CREATE VIEW reporting.customers AS
SELECT
    customer_id,
    full_name,
    email,
    country,
    created_at,
    subscription_plan
FROM silver.customers;
-- Новые колонки в базовой таблице НЕ попадут во view автоматически
-- View стабильна до тех пор, пока существующие колонки не изменятся

Явное перечисление колонок - страховка от двух сторон:

Новые колонки в таблице не "протекают" во view неожиданно
Попытка добавить новую колонку в таблицу с SELECT * во view может дать сюрпризы downstream-потребителям

Приоритет имён: Temp View vs Persistent Table¶

Что произойдёт если Temp View и Persistent Table имеют одинаковое имя?

# В metastore существует таблица silver.users
# Создаём Temp View с тем же именем
spark.sql("""
    CREATE OR REPLACE TEMPORARY VIEW users AS
    SELECT user_id, 'REDACTED' AS email FROM silver.users
""")

# Какой users будет использован?
spark.sql("SELECT * FROM users LIMIT 5").show()
# → Temp View выигрывает! email = 'REDACTED'

# Чтобы явно обратиться к таблице из metastore:
spark.sql("SELECT * FROM silver.users LIMIT 5").show()
# → реальная таблица с настоящим email

Правило приоритета: Temp View > Persistent Table с тем же именем.

Это поведение - и инструмент, и ловушка:

Как инструмент: можно создать Temp View с именем orders который фильтрует данные только для текущего пользователя, и все SQL-запросы этой сессии будут автоматически видеть только "свои" данные.

Как ловушка: разработчик создаёт Temp View payments для тестирования и забывает о нём. Все запросы в этой сессии теперь читают тестовые данные вместо production-таблицы. Ошибку сложно обнаружить.

Проверка конфликтов имён¶

def check_name_conflicts(spark: SparkSession) -> None:
    """Находит конфликты имён между Temp Views и Persistent Tables."""
    temp_views = {
        t.name
        for t in spark.catalog.listTables()
        if t.isTemporary
    }
    persistent = {
        t.name
        for t in spark.catalog.listTables()
        if not t.isTemporary
    }
    conflicts = temp_views & persistent
    if conflicts:
        print(f"⚠ Name conflicts (Temp View shadows Persistent Table):")
        for name in conflicts:
            print(f"  - {name}")
    else:
        print("✓ No name conflicts")


check_name_conflicts(spark)

Views как Security Abstraction¶

Persistent Views - мощный инструмент для скрытия чувствительных данных и реализации row-level security без изменения базовых таблиц.

Маскирование PII-колонок¶

-- Базовая таблица содержит PII: email, phone, name
-- silver.customers - только для data engineers

-- Публичная view для аналитиков: PII замаскировано
CREATE OR REPLACE VIEW reporting.customers_masked AS
SELECT
    customer_id,
    -- Маскируем email: оставляем домен, скрываем username
    CONCAT(
        REPEAT('*', 4),
        '@',
        SPLIT(email, '@')[1]
    )                           AS email_masked,
    -- Маскируем телефон: оставляем код страны + XXX
    CONCAT(SUBSTRING(phone, 1, 3), '****', SUBSTRING(phone, -4))
                                AS phone_masked,
    country,
    created_at,
    subscription_plan,
    total_orders,
    total_spent
FROM silver.customers;

-- Аналитики работают с reporting.customers_masked
-- и никогда не видят реальные email и телефон

Row-level Security через View¶

-- Продажи по регионам: каждый менеджер видит только свой регион
CREATE OR REPLACE VIEW reporting.my_region_sales AS
SELECT *
FROM gold.sales
WHERE region = (
    -- current_user() возвращает имя текущего пользователя Spark
    SELECT region
    FROM security.user_regions
    WHERE username = CURRENT_USER()
);

Фильтрация тестовых данных¶

-- silver.events содержит реальные и тестовые события
-- Analytics view фильтрует тестовые

CREATE OR REPLACE VIEW analytics.clean_events AS
SELECT *
FROM silver.events
WHERE is_test = FALSE
  AND user_id IS NOT NULL
  AND user_id NOT IN (SELECT user_id FROM security.bot_users)
  AND environment = 'production';

Практика: semantic layer для BI¶

Построим полноценный аналитический слой из Persistent Views поверх Silver-таблиц.

Шаг 1: Silver-таблицы (основа)¶

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, current_timestamp, lit
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("views-practice") \
    .enableHiveSupport() \
    .getOrCreate()

# Предположим, Silver-таблицы уже существуют:
# silver.orders(order_id, customer_id, order_date, status, total_amount, is_test)
# silver.order_items(item_id, order_id, sku, category, qty, line_total)
# silver.customers(customer_id, name, email, country, subscription_plan)

Шаг 2: Создание reporting-базы данных и Views¶

spark.sql("CREATE DATABASE IF NOT EXISTS reporting COMMENT 'Semantic layer for BI and Analytics'")

# View 1: чистые заказы (без тестовых, без отменённых)
spark.sql("""
    CREATE OR REPLACE VIEW reporting.valid_orders AS
    SELECT
        order_id,
        customer_id,
        order_date,
        status,
        total_amount
    FROM silver.orders
    WHERE is_test = FALSE
      AND status NOT IN ('cancelled', 'test_order')
""")

# View 2: клиенты с маскированным email (PII protection)
spark.sql("""
    CREATE OR REPLACE VIEW reporting.customers AS
    SELECT
        customer_id,
        name,
        CONCAT(REPEAT('*', 4), '@', SPLIT(email, '@')[1]) AS email,
        country,
        subscription_plan
    FROM silver.customers
""")

# View 3: KPI витрина - метрики на уровне заказа
spark.sql("""
    CREATE OR REPLACE VIEW reporting.order_metrics AS
    SELECT
        o.order_id,
        o.customer_id,
        o.order_date,
        o.total_amount,
        COUNT(oi.item_id)            AS item_count,
        SUM(oi.qty)                  AS total_units,
        COUNT(DISTINCT oi.category)  AS categories_count
    FROM reporting.valid_orders o
    JOIN silver.order_items oi USING (order_id)
    GROUP BY o.order_id, o.customer_id, o.order_date, o.total_amount
""")

Шаг 3: Аналитические запросы через views¶

# Аналитик 1: топ-клиенты по выручке
top_customers = spark.sql("""
    SELECT
        c.customer_id,
        c.name,
        c.country,
        c.subscription_plan,
        COUNT(om.order_id)      AS total_orders,
        SUM(om.total_amount)    AS total_revenue,
        AVG(om.item_count)      AS avg_items_per_order
    FROM reporting.order_metrics om
    JOIN reporting.customers c USING (customer_id)
    WHERE om.order_date >= DATE_SUB(CURRENT_DATE(), 30)
    GROUP BY c.customer_id, c.name, c.country, c.subscription_plan
    ORDER BY total_revenue DESC
    LIMIT 20
""")
top_customers.show(truncate=False)

# Аналитик 2: ежедневная выручка по категориям
daily_revenue = spark.sql("""
    SELECT
        o.order_date,
        oi.category,
        SUM(oi.line_total)           AS revenue,
        COUNT(DISTINCT o.order_id)   AS orders
    FROM reporting.valid_orders o
    JOIN silver.order_items oi USING (order_id)
    GROUP BY o.order_date, oi.category
    ORDER BY o.order_date DESC, revenue DESC
""")
daily_revenue.show(10)

Шаг 4: Демонстрация double evaluation и его исправление¶

# Запрос с двойным обращением к reporting.order_metrics
# (view поверх JOIN + GROUP BY → дорогая операция)
problematic = spark.sql("""
    SELECT
        om.customer_id,
        om.total_amount,
        avg_stats.avg_order
    FROM reporting.order_metrics om
    CROSS JOIN (
        SELECT AVG(total_amount) AS avg_order
        FROM reporting.order_metrics   ← ВТОРОЙ раз!
    ) avg_stats
    WHERE om.total_amount > avg_stats.avg_order
""")

# В explain увидим два одинаковых поддерева
problematic.explain(mode="formatted")
# Ищем два FileScan silver.orders + два FileScan silver.order_items

# ИСПРАВЛЕНИЕ: кэшируем view
spark.sql("CACHE TABLE reporting.order_metrics")

# Принудительно материализуем (важно: CACHE TABLE - lazy!)
spark.sql("SELECT COUNT(*) FROM reporting.order_metrics").show()

# Теперь plan использует InMemoryRelation
fixed = spark.sql("""
    SELECT om.customer_id, om.total_amount, avg_stats.avg_order
    FROM reporting.order_metrics om
    CROSS JOIN (
        SELECT AVG(total_amount) AS avg_order
        FROM reporting.order_metrics
    ) avg_stats
    WHERE om.total_amount > avg_stats.avg_order
""")
fixed.explain(mode="formatted")
# → InMemoryRelation (один кэш, два чтения из него)

# Освобождаем
spark.sql("UNCACHE TABLE reporting.order_metrics")

Шаг 5: Инспекция Views через catalog¶

# Все views в reporting
print("Views in reporting:")
for table in spark.catalog.listTables("reporting"):
    print(f"  {table.name} ({table.tableType})")

# Просмотреть определение view
create_stmt = spark.sql("SHOW CREATE TABLE reporting.valid_orders").collect()[0]["createtab_stmt"]
print("\nView definition:")
print(create_stmt)

# Описание view (как таблицы)
spark.sql("DESCRIBE TABLE EXTENDED reporting.valid_orders").show(truncate=False)
# В выводе: Type = VIEW, View Text = SELECT ...

View vs CTE: когда что выбрать¶

CTE (WITH clause) - альтернатива view для query-scoped абстракций:

-- CTE: живёт только внутри этого запроса
WITH
clean_orders AS (
    SELECT * FROM silver.orders
    WHERE status = 'completed' AND is_test = FALSE
),
order_totals AS (
    SELECT order_id, SUM(line_total) AS total
    FROM silver.order_items
    GROUP BY order_id
)
SELECT o.customer_id, SUM(t.total) AS revenue
FROM clean_orders o
JOIN order_totals t USING (order_id)
GROUP BY o.customer_id;

Критерий	CTE	Temp View	Persistent View
Область видимости	Один запрос	Одна сессия	Все приложения
Переиспользование	Нет	В рамках сессии	Долгосрочное
Возможность кэширования	Нет	CACHE TABLE	CACHE TABLE
Поддержка в Spark UI	Inline в плане	Как отдельный план	Как отдельный план
Подходит для	Декомпозиция длинного запроса	Ad-hoc аналитика	Семантический слой

Практическое правило:

CTE: когда абстракция нужна только внутри одного запроса
Temp View: когда нужно использовать результат в нескольких запросах одной сессии
Persistent View: когда бизнес-логика должна быть общей для всей команды и всех инструментов

Debugging Views: полезные команды¶

-- Просмотреть определение view
SHOW CREATE TABLE reporting.valid_orders;

-- Развернуть plan для отладки вложенности
EXPLAIN EXTENDED SELECT * FROM reporting.order_metrics LIMIT 1;

-- Список всех views (тип VIEW) в базе данных
SHOW TABLES IN reporting;
-- isTemporary = false → persistent view или table

-- Описание view включает "View Text"
DESCRIBE TABLE EXTENDED reporting.valid_orders;

# Programmatic: тип объекта
for t in spark.catalog.listTables("reporting"):
    print(f"{t.name}: isTemporary={t.isTemporary}, type={t.tableType}")
    # VIEW: isTemporary=False, type=VIEW

# Проверка что view существует
spark.catalog.tableExists("reporting.valid_orders")     # → True
spark.catalog.tableExists("reporting.nonexistent_view") # → False

# Удалить конкретную temp view
spark.catalog.dropTempView("active_orders")

# Удалить все temp views (в Spark 3.1+)
spark.catalog.clearCache()  # очищает кэш
# Temp views удаляются только вместе с сессией или через dropTempView

Anti-patterns¶

Anti-pattern 1: SELECT * во Persistent View¶

-- ПЛОХО: SELECT * делает view нестабильной
CREATE VIEW reporting.orders AS
SELECT * FROM silver.orders;
-- При добавлении новой колонки в silver.orders она протечёт в reporting.orders
-- BI-инструменты могут получить неожиданные данные

-- ХОРОШО: явный список
CREATE VIEW reporting.orders AS
SELECT order_id, customer_id, order_date, status, total_amount
FROM silver.orders;

Anti-pattern 2: Глубокая вложенность Views¶

-- ПЛОХО: 5+ уровней вложенности
CREATE VIEW v5 AS SELECT * FROM v4 WHERE ...;
CREATE VIEW v4 AS SELECT * FROM v3 JOIN t ON ...;
CREATE VIEW v3 AS SELECT * FROM v2 WHERE ...;
CREATE VIEW v2 AS SELECT * FROM v1 GROUP BY ...;
CREATE VIEW v1 AS SELECT * FROM silver.raw WHERE ...;
-- Catalyst будет строить гигантский план, debugging невозможен

-- ХОРОШО: максимум 2-3 уровня, промежуточные результаты материализуем
-- silver.raw → silver.clean (Delta table) → reporting.kpi (view 1 уровень)

Anti-pattern 3: Не очищать кэш после использования¶

# ПЛОХО: кэш занимает память, не освобождается
spark.sql("CACHE TABLE heavy_view")
result = spark.sql("SELECT * FROM heavy_view")
result.write.parquet("output/")
# Забыли UNCACHE - heavy_view занимает memory до конца сессии

# ХОРОШО: явное освобождение
try:
    spark.sql("CACHE TABLE heavy_view")
    spark.sql("SELECT COUNT(*) FROM heavy_view").show()  # trigger
    result = spark.sql("SELECT * FROM heavy_view ...")
    result.write.parquet("output/")
finally:
    spark.sql("UNCACHE TABLE heavy_view")

Anti-pattern 4: Persistent View поверх Temp View¶

# НЕПРАВИЛЬНО: попытка создать permanent view на temp view
spark.sql("CREATE TEMP VIEW session_data AS SELECT ...")
spark.sql("CREATE VIEW reporting.stable_view AS SELECT * FROM session_data")
# → AnalysisException: Cannot create permanent view on temp view

Anti-pattern 5: Не использовать IF NOT EXISTS / OR REPLACE¶

-- СЛОМАЕТСЯ при повторном запуске:
CREATE VIEW reporting.customers AS ...;
-- AnalysisException: View `reporting`.`customers` already exists

-- БЕЗОПАСНО:
CREATE OR REPLACE VIEW reporting.customers AS ...;
-- или
CREATE VIEW IF NOT EXISTS reporting.customers AS ...;

Домашнее задание¶

Дан ETL-скрипт с пятью Temp Views:

# Входные данные
spark.sql("CREATE TEMP VIEW raw_events AS SELECT * FROM silver.events")

spark.sql("""
    CREATE TEMP VIEW clean_events AS
    SELECT * FROM raw_events
    WHERE user_id IS NOT NULL AND is_test = FALSE
""")

spark.sql("""
    CREATE TEMP VIEW user_session_stats AS
    SELECT user_id, session_id, COUNT(*) AS event_count, SUM(revenue) AS session_revenue
    FROM clean_events
    GROUP BY user_id, session_id
""")

spark.sql("""
    CREATE TEMP VIEW user_total_stats AS
    SELECT user_id, COUNT(*) AS sessions, SUM(session_revenue) AS total_revenue
    FROM user_session_stats
    GROUP BY user_id
""")

spark.sql("""
    CREATE TEMP VIEW high_value_users AS
    SELECT * FROM user_total_stats
    WHERE total_revenue > (SELECT AVG(total_revenue) FROM user_total_stats)
""")

# Финальный запрос - медленный
result = spark.sql("""
    SELECT u.user_id, u.sessions, u.total_revenue, p.country
    FROM high_value_users u
    JOIN silver.profiles p USING (user_id)
    ORDER BY u.total_revenue DESC
""")
result.show(20)

Задания:

Запустить скрипт и открыть Spark UI. Найти stage с двойным вычислением через анализ DAG-графа. Определить какая view вычисляется дважды и почему
Применить точечное кэширование: CACHE TABLE только для той view, которая создаёт проблему. Сравнить время выполнения до и после. Сравнить explain-планы
Переписать весь пайплайн используя только CTE (WITH clause) без Temp Views. Сравнить читаемость и время выполнения
Создать Persistent View analytics.high_value_user_summary которая:
Использует явный список колонок (не SELECT *)
Маскирует поле email из silver.profiles
Фильтрует is_test = FALSE
Написать функцию validate_view_chain(spark, views_list) которая проходит по списку view и проверяет каждую на корректность (не упадёт ли при выполнении). Вернуть отчёт: какие view в порядке, какие сломаны и с какой ошибкой

Чеклист¶

[ ] Понимаю три типа Views: Temp, Global Temp, Persistent - время жизни и область видимости каждого
[ ] Знаю разницу createOrReplaceTempView() vs createOrReplaceGlobalTempView() - где и как доступен результат
[ ] Знаю что Global Temp View живёт в базе global_temp и требует явного префикса global_temp.view_name
[ ] Понимаю View Expansion: Catalyst подставляет тело view в план запроса, нет специального оператора "View" в физическом плане
[ ] Знаю что predicate pushdown и column pruning работают сквозь view-абстракцию
[ ] Понимаю ловушку Double Evaluation: один view используется дважды → два раза вычисляется весь запрос
[ ] Умею обнаружить double evaluation через explain: ищу два одинаковых поддерева в Physical Plan
[ ] Знаю три решения double evaluation: CTE, CACHE TABLE, DataFrame.persist()
[ ] Понимаю Late Binding: схема view проверяется при выполнении, не при создании
[ ] Знаю что ALTER TABLE (rename/drop column) может сломать dependent views
[ ] Никогда не пишу SELECT * в Persistent View - только явный список колонок
[ ] Знаю приоритет имён: Temp View затеняет Persistent Table с тем же именем
[ ] Умею освобождать кэш через UNCACHE TABLE / unpersist()
[ ] Знаю ограничение: Persistent View нельзя создать поверх Temp View
[ ] Умею использовать Views для PII-маскирования и row-level filtering
[ ] Знаю разницу View vs CTE - когда что использовать