Apache Spark 4 у 2026 році: нові можливості, Structured Streaming та питання для співбесіди

Apache Spark 4 є найбільш значним оновленням з часів версії 3.0, що вносить фундаментальні зміни в екосистему обробки даних. Завдяки активації режиму ANSI SQL за замовчуванням, новому типу даних VARIANT для напівструктурованих даних та повністю переробленому фреймворку стримінгу Spark 4 позиціонується як центральна платформа для побудови сучасних конвеєрів обробки даних. У цій статті проаналізовано ключові нововведення, продемонстровано їхнє практичне застосування на конкретних прикладах коду та підготовлено цільовий набір технічних питань для співбесід у галузі Data Engineering.

Режим ANSI SQL як новий стандарт

Одна з найбільш масштабних змін у Spark 4 стосується активації режиму ANSI SQL за замовчуванням. У попередніх версіях некоректні перетворення типів мовчки повертали значення NULL, тоді як Spark 4 тепер генерує виняток ArithmeticException. Ця зміна поведінки суттєво покращує якість даних, проте вимагає систематичної міграції існуючих конвеєрів.

Найнадійніший шлях переходу полягає у використанні TRY_CAST, який зберігає попередню семантику NULL і водночас працює відповідно до стандарту ANSI.

# ansi_mode_migration.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, try_cast

spark = SparkSession.builder.appName("ANSIMigration").getOrCreate()

# This now throws ArithmeticException in Spark 4 ANSI mode
# SELECT CAST('abc' AS INT)  -- runtime error

# Safe migration: use TRY_CAST to preserve NULL behavior
df = spark.sql("""
    SELECT
        TRY_CAST(raw_value AS INT) AS parsed_value,
        TRY_CAST(amount AS DECIMAL(10,2)) AS safe_amount
    FROM raw_events
    WHERE TRY_CAST(raw_value AS INT) IS NOT NULL
""")

# Alternatively, wrap arithmetic in try_divide / try_add
result = df.withColumn(
    "ratio",
    col("safe_amount") / col("parsed_value")  # throws on zero
)

Команди, які мігрують великі кодові бази, повинні спочатку провести повний аудит, під час якого всі вирази CAST будуть ідентифіковані та замінені на TRY_CAST. На практиці перевіреним підходом є попередня активація режиму ANSI у середовищі staging із систематичним протоколюванням винятків, перш ніж зміна буде впроваджена у продакшн.

Тип даних VARIANT для напівструктурованих даних

Новий тип даних VARIANT вирішує давню проблему в екосистемі Spark: ефективну обробку напівструктурованих даних, таких як JSON, Avro або Protobuf. Раніше робота з вкладеними структурами JSON вимагала або складних визначень схем, або обхідних шляхів через парсинг рядків. VARIANT зберігає дані в оптимізованому бінарному форматі з автоматичним shredding, завдяки чому доступ до вкладених полів відбувається значно швидше, ніж при традиційному парсингу JSON.

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder.appName("VariantDemo").getOrCreate()
import spark.implicits._

// Create a table with VARIANT column
spark.sql("""
    CREATE TABLE events (
        event_id BIGINT,
        timestamp TIMESTAMP,
        payload VARIANT
    ) USING PARQUET
""")

// Insert JSON data — automatically converted to VARIANT
spark.sql("""
    INSERT INTO events VALUES
    (1, current_timestamp(),
     PARSE_JSON('{"user_id": 42, "action": "click", "metadata": {"page": "/home", "duration_ms": 320}}')),
    (2, current_timestamp(),
     PARSE_JSON('{"user_id": 87, "action": "purchase", "metadata": {"item_id": "SKU-9001", "amount": 49.99}}'))
""")

// Query nested fields with dot notation — uses shredding for fast access
val clicks = spark.sql("""
    SELECT
        event_id,
        payload:user_id::INT AS user_id,
        payload:action::STRING AS action,
        payload:metadata.page::STRING AS page
    FROM events
    WHERE payload:action::STRING = 'click'
""")
clicks.show()

Головна перевага VARIANT порівняно з класичними STRING-стовпцями з from_json полягає в ефективності зберігання та швидкості запитів. Завдяки автоматичному shredding Spark витягує часто запитувані поля в окремі стовпці, тоді як рідше використовувані дані зберігаються у стисненому вигляді. У бенчмарках VARIANT демонструє до 10-кратного прискорення при типових запитах до вкладених JSON-даних порівняно з обробкою на основі рядків.

Для Data Engineers це означає, що дані подій з Kafka, REST API або IoT-пристроїв можна записувати безпосередньо в таблиці Spark без попереднього визначення схеми, а потім запитувати за допомогою SQL-синтаксису.

Real-Time Mode Streaming у Spark 4.1

Spark 4.1 запроваджує Real-Time Mode як нову опцію стримінгу, що знижує затримку обробки до рівня нижче однієї секунди. Якщо попередня модель Micro-Batch обробляла дані дискретними інтервалами, то Real-Time Mode працює безперервно та негайно передає результати до наступних систем у ланцюзі.

Конфігурація здійснюється через параметр spark.sql.streaming.mode, який встановлюється у значення realtime.

# real_time_streaming.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window
from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

spark = SparkSession.builder \
    .appName("RTMDemo") \
    .config("spark.sql.streaming.mode", "realtime") \
    .getOrCreate()

schema = StructType([
    StructField("sensor_id", StringType()),
    StructField("temperature", DoubleType()),
    StructField("event_time", TimestampType())
])

# Read from Kafka with Real-Time Mode enabled
raw_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "sensor-readings") \
    .option("startingOffsets", "latest") \
    .load()

parsed = raw_stream \
    .select(from_json(col("value").cast("string"), schema).alias("data")) \
    .select("data.*")

# Write to Kafka sink — sub-second end-to-end
query = parsed \
    .filter(col("temperature") > 80.0) \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("topic", "alerts") \
    .option("checkpointLocation", "/tmp/checkpoints/alerts") \
    .start()

query.awaitTermination()

Real-Time Mode особливо підходить для таких сценаріїв, як виявлення шахрайства, алертинг у реальному часі та операційні дашборди, де кожна секунда затримки може мати критичні бізнес-наслідки. Водночас слід враховувати, що цей режим висуває вищі вимоги до ресурсів порівняно з традиційною моделлю Micro-Batch, тому його варто використовувати лише у випадках, коли затримка менше секунди справді необхідна.

Spark Connect та легковагий клієнт PySpark

Spark Connect повністю відокремлює клієнт від кластера та дає змогу виконувати операції Spark через тонку клієнтську бібліотеку без локальної JVM. Це архітектурне рішення суттєво спрощує розгортання у контейнеризованих середовищах, оскільки розробники та Data Scientists можуть запускати Spark-джоби безпосередньо з Jupyter Notebooks або CI/CD-конвеєрів, не маючи повної інсталяції Spark на локальній машині.

# spark_connect_client.py
from pyspark.sql import SparkSession

# Connect to remote Spark cluster — no local JVM needed
spark = SparkSession.builder \
    .remote("sc://spark-cluster.internal:15002") \
    .getOrCreate()

# Full DataFrame API available through thin client
df = spark.read.parquet("s3a://data-lake/events/2026/")

aggregated = df.groupBy("country", "event_type") \
    .count() \
    .orderBy("count", ascending=False)

aggregated.show(20)

Протокол Spark Connect базується на gRPC і серіалізує логічні плани запитів замість переміщення даних між клієнтом і сервером. Це призводить до суттєвого зменшення мережевих витрат та уможливлює сценарії мультитенантності, коли сотні користувачів одночасно працюють з одним кластером. Для підприємств це означає кращу утилізацію ресурсів та спрощене управління доступом через стандартизовані ендпоінти.

API transformWithState для складної стримінгової логіки

Новий API transformWithState замінює застарілі mapGroupsWithState та flatMapGroupsWithState типобезпечним, композиційним інтерфейсом. Ключова перевага полягає в явному управлінні станом з TTL (Time-to-Live), що забезпечує автоматичне очищення неактивних сесій та вирішує поширену проблему у продакшн-стримінгових конвеєрах.

import org.apache.spark.sql.streaming._
import org.apache.spark.sql.streaming.StatefulProcessor
import java.time.Duration

case class UserEvent(userId: String, action: String, timestamp: Long)
case class SessionSummary(userId: String, actionCount: Int, durationMs: Long)

class SessionProcessor extends StatefulProcessor[String, UserEvent, SessionSummary] {

  // ValueState with 30-minute TTL — auto-cleanup of idle sessions
  @transient private var sessionStart: ValueState[Long] = _
  @transient private var actionCount: ValueState[Int] = _

  override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
    sessionStart = getHandle.getValueState[Long]("start", TTLConfig(Duration.ofMinutes(30)))
    actionCount = getHandle.getValueState[Int]("count", TTLConfig(Duration.ofMinutes(30)))
  }

  override def handleInputRows(
      key: String,
      rows: Iterator[UserEvent],
      timerValues: TimerValues
  ): Iterator[SessionSummary] = {
    rows.flatMap { event =>
      if (!sessionStart.exists()) {
        sessionStart.update(event.timestamp)
        actionCount.update(1)
        Iterator.empty
      } else {
        val count = actionCount.get() + 1
        actionCount.update(count)
        // Emit summary every 10 actions
        if (count % 10 == 0) {
          val duration = event.timestamp - sessionStart.get()
          Iterator(SessionSummary(key, count, duration))
        } else Iterator.empty
      }
    }
  }
}

Конфігурація TTL гарантує, що стан неактивних сесій користувачів буде автоматично очищено через 30 хвилин. Без цього механізму State Store у довготривалих стримінгових додатках безперервно зростав би й зрештою призводив до проблем із пам'яттю. На співбесідах часто ставлять запитання саме про цей аспект, оскільки він демонструє розуміння операційних викликів стримінгових систем.

Spark Declarative Pipelines (SDP) у Spark 4.1

Spark Declarative Pipelines (SDP) дають змогу визначати ETL-процеси декларативним способом. Замість ручної оркестрації плану виконання розробник лише описує бажані таблиці та їхні залежності. Spark автоматично створює DAG виконання, паралелізує обробку та керує обробкою помилок.

# declarative_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as _sum, current_date

spark = SparkSession.builder.appName("SDP-Demo").getOrCreate()

# Define datasets declaratively — Spark resolves dependencies
@spark.declarative.table("clean_orders")
def clean_orders():
    return spark.read.table("raw_orders") \
        .filter(col("status") != "cancelled") \
        .filter(col("amount") > 0)

@spark.declarative.table("daily_revenue")
def daily_revenue():
    return spark.read.table("clean_orders") \
        .groupBy("order_date", "region") \
        .agg(_sum("amount").alias("total_revenue"))

@spark.declarative.table("revenue_summary")
def revenue_summary():
    return spark.read.table("daily_revenue") \
        .groupBy("region") \
        .agg(_sum("total_revenue").alias("grand_total")) \
        .orderBy("grand_total", ascending=False)

# Execute — Spark builds the DAG, parallelizes, handles retries
spark.declarative.run()

Цей підхід концептуально нагадує dbt, проте працює безпосередньо у середовищі виконання Spark. Це усуває необхідність зовнішніх інструментів оркестрації для простих та середніх конвеєрів та зменшує загальну складність стеку Data Engineering. Особливо цінною є автоматична розв'язка залежностей: коли revenue_summary залежить від daily_revenue, а та, у свою чергу, від clean_orders, Spark самостійно обчислює оптимальний порядок виконання.

Питання для співбесіди з Data Engineering щодо Spark 4

Наведені нижче питання регулярно з'являються на технічних співбесідах на позиції Data Engineering та охоплюють ключові нововведення Spark 4.

Питання 1: Чому Spark 4 активував режим ANSI SQL за замовчуванням і як це впливає на існуючі конвеєри?

Режим ANSI SQL покращує якість даних, оскільки некоректні перетворення типів та арифметичні переповнення більше не обробляються мовчки як NULL, а генерують явні винятки. Існуючі конвеєри мають замінити всі вирази CAST на TRY_CAST для збереження попередньої поведінки. Ця зміна виявляє приховані помилки в даних, які раніше непомітно поширювалися через конвеєр.

Питання 2: Що таке тип даних VARIANT і чим він відрізняється від обробки JSON на основі STRING?

VARIANT зберігає напівструктуровані дані в оптимізованому бінарному форматі з автоматичним Column Shredding. На відміну від STRING-стовпців із from_json, VARIANT не вимагає визначення схеми при збереженні, але при цьому забезпечує типізований доступ через нотацію шляху. Швидкість запитів до 10 разів вища, оскільки часто запитувані поля автоматично матеріалізуються як окремі стовпці.

Питання 3: Чим Real-Time Mode відрізняється від попередньої моделі Micro-Batch у Structured Streaming?

Модель Micro-Batch обробляє дані дискретними часовими інтервалами з типовими затримками від 100 мс до кількох секунд. Real-Time Mode обробляє кожен запис одразу після надходження та досягає затримок менше секунди. Режим підходить для критичних до часу застосувань, таких як виявлення шахрайства, проте вимагає більше ресурсів і наразі підтримує лише сценарії Kafka-до-Kafka.

Питання 4: Які переваги надає Spark Connect порівняно з традиційним клієнтським режимом?

Spark Connect повністю відокремлює клієнт від кластера через інтерфейс gRPC. Клієнт не потребує локальної JVM, що спрощує інтеграцію в контейнерні середовища та блокноти. Логічні плани запитів серіалізуються та передаються, а не самі дані, що мінімізує мережеві витрати. Декілька користувачів можуть використовувати один кластер через стандартизовані ендпоінти.

Питання 5: Яку проблему вирішує API transformWithState і чому його було впроваджено?

API transformWithState замінює mapGroupsWithState та flatMapGroupsWithState типобезпечним інтерфейсом із нативною підтримкою TTL. Він вирішує проблему необмеженого зростання стану у довготривалих стримінгових додатках завдяки автоматичному очищенню неактивних станів. API надає можливість композиційних визначень стану з ValueState, ListState та MapState.

Повний ETL-конвеєр на Spark 4: Bronze-Silver-Gold

Наступний приклад демонструє продакшн-готовий ETL-конвеєр, що інтегрує всі нові можливості Spark 4: режим ANSI SQL, тип даних VARIANT та Structured Streaming з Kafka як джерелом і Delta Lake як приймачем.

# spark4_etl_pipeline.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, current_timestamp, parse_json

spark = SparkSession.builder \
    .appName("Spark4ETL") \
    .config("spark.sql.ansi.enabled", "true") \
    .getOrCreate()

# Bronze: raw ingestion from Kafka
bronze = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "user-events") \
    .load() \
    .select(
        col("key").cast("string").alias("event_key"),
        parse_json(col("value").cast("string")).alias("payload"),
        col("timestamp").alias("kafka_ts")
    )

# Silver: extract typed fields from VARIANT
silver = bronze.select(
    col("event_key"),
    col("payload:user_id").cast("int").alias("user_id"),
    col("payload:action").cast("string").alias("action"),
    col("payload:metadata.session_id").cast("string").alias("session_id"),
    col("kafka_ts"),
    current_timestamp().alias("processed_at")
).filter(
    col("user_id").isNotNull()  # ANSI mode makes this explicit
)

# Gold: write to Delta Lake with schema enforcement
query = silver.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3a://checkpoints/user-events/") \
    .option("mergeSchema", "true") \
    .toTable("gold.user_events")

query.awaitTermination()

Шар Bronze приймає необроблені дані з Kafka та конвертує їх за допомогою parse_json безпосередньо у тип даних VARIANT. У шарі Silver відбувається типізована екстракція окремих полів через нотацію шляху. Режим ANSI гарантує, що помилкові перетворення будуть виявлені негайно, замість потрапляння у шар Gold як значення NULL. Шар Gold записує очищені та типізовані дані у Delta Lake з активованою еволюцією схеми.

Ця архітектура представляє актуальний підхід найкращих практик для стримінгового ETL на Spark 4 і поєднує всі суттєві нововведення в узгоджену систему.

Висновок

Apache Spark 4 знаменує зміну поколінь в обробці великих даних. Нововведення адресують конкретні слабкі місця попередніх версій та позиціонують Spark як комплексну платформу для пакетної обробки та обробки в реальному часі одночасно.

Основні досягнення в огляді:

• Режим ANSI SQL за замовчуванням покращує якість даних завдяки явній обробці помилок при перетвореннях типів
• Тип даних VARIANT спрощує роботу з напівструктурованими даними при значно вищій швидкості запитів
• Real-Time Mode Streaming знижує наскрізну затримку до рівня нижче секунди для критичних до часу сценаріїв
• Spark Connect розділяє клієнт і кластер та дає змогу здійснювати легковагу інтеграцію без локальної JVM
• API transformWithState пропонує типобезпечне управління станом з автоматичним очищенням на основі TTL
• Spark Declarative Pipelines зменшують складність ETL-процесів завдяки автоматичній розв'язці залежностей

Для Data Engineers та розробників, які готуються до технічних співбесід, знання цих можливостей стає незамінним. Здатність пояснити відмінності між Micro-Batch та Real-Time Mode, продемонструвати переваги VARIANT порівняно з обробкою JSON на основі STRING та побудувати повний конвеєр Bronze-Silver-Gold на Spark 4 формує основу для успішних співбесід у галузі Data Engineering.