Apache Spark 4 w 2026 roku: Nowe funkcje, Structured Streaming i pytania rekrutacyjne

Apache Spark 4 stanowi najbardziej znaczace wydanie od czasu wersji 3.0, wprowadzajac fundamentalne zmiany w calym ekosystemie Data Engineering. Wlaczenie trybu ANSI SQL jako domyslnego standardu, nowy typ danych VARIANT do obslugi danych polstrukturalnych oraz gruntownie przebudowany framework streamingowy pozycjonuja Spark 4 jako centralna platforme nowoczesnych potoków przetwarzania danych. Niniejszy artykul analizuje najwazniejsze nowosci, demonstruje ich praktyczne zastosowanie na konkretnych przykladach kodu i przygotowuje do pytan technicznych na rozmowach kwalifikacyjnych w obszarze Data Engineering.

Tryb ANSI SQL jako nowy standard

Jedna z najdalej idacych zmian w Spark 4 dotyczy wlaczenia trybu ANSI SQL jako domyslnej konfiguracji. Podczas gdy wczesniejsze wersje przy blednych konwersjach typów zwracaly po cichu wartosc NULL, Spark 4 rzuca teraz wyjatek ArithmeticException. Ta zmiana zachowania znaczaco poprawia jakosc danych, wymaga jednak systematycznej migracji istniejacych potoków przetwarzania.

Najbezpieczniejsza sciezka migracji polega na zastosowaniu TRY_CAST, które zachowuje dotychczasowa semantyke NULL i jednoczesnie jest zgodne ze standardem 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
)

Zespoly migrujace duze bazy kodu powinny zaplanowac pelna faze audytu, w ramach której wszystkie wyrazenia CAST zostana zidentyfikowane i zastapione przez TRY_CAST. W praktyce sprawdzonym podejsciem jest najpierw aktywacja trybu ANSI w srodowiskach staging i systematyczne rejestrowanie powstalych wyjatkow, zanim zmiana zostanie wdrozona na produkcji.

Typ danych VARIANT dla danych polstrukturalnych

Nowy typ danych VARIANT rozwiazuje dlugotrwaly problem w swiecie Spark: wydajne przetwarzanie danych polstrukturalnych, takich jak JSON, Avro czy Protobuf. Dotychczas praca z zagniezdonymi strukturami JSON wymagala albo skomplikowanych definicji schematów, albo obejscia przez parsowanie lancuchów znakowych. VARIANT przechowuje dane w zoptymalizowanym formacie binarnym z automatycznym shredding, dzieki czemu dostep do zagniezdonych pól jest znacznie szybszy niz przy tradycyjnym parsowaniu 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()

Kluczowa przewaga VARIANT nad klasycznymi kolumnami STRING z from_json polega na efektywnosci przechowywania i szybkosci zapytan. Dzieki automatycznemu shredding Spark wyodrebnia czesto odpytywane pola do osobnych kolumn, podczas gdy rzadziej potrzebne dane pozostaja skompresowane. W testach wydajnosciowych VARIANT wykazuje nawet 10-krotne przyspieszenie przy typowych zapytaniach na zagniezdonych danych JSON w porównaniu z przetwarzaniem opartym na lancuchach znakowych.

Dla inzynierow danych oznacza to, ze dane zdarzen z Kafki, REST API czy urzadzen IoT moga byc zapisywane bezposrednio do tabel Spark bez uprzedniej definicji schematu, a nastepnie odpytywane za pomoca skladni SQL.

Real-Time Mode Streaming w Spark 4.1

Spark 4.1 wprowadza Real-Time Mode jako nowa opcje streamingu, obniajaca opóznienie przetwarzania ponizej jednej sekundy. Podczas gdy dotychczasowy model Micro-Batch przetwarza dane w dyskretnych przedzialach czasowych, Real-Time Mode dziala w sposób ciagly i natychmiast przekazuje wyniki do systemów docelowych.

Konfiguracja odbywa sie za pomoca parametru spark.sql.streaming.mode, który nalezy ustawic na 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 jest szczególnie odpowiedni dla przypadków uzycia takich jak wykrywanie oszustw, alertowanie w czasie rzeczywistym oraz operacyjne dashboardy, gdzie kazda sekunda opóznienia moze miec krytyczne konsekwencje biznesowe. Nalezy jednak pamietac, ze ten tryb stawia wyzsze wymagania zasobowe niz tradycyjny model Micro-Batch i powinien byc stosowany wylacznie w scenariuszach, gdzie opóznienie ponizej sekundy jest faktycznie wymagane.

Spark Connect i lekki klient PySpark

Spark Connect calkowicie oddziela klienta od klastra, umozliwiajac wykonywanie operacji Spark poprzez cienka biblioteke kliencka bez lokalnej maszyny JVM. Ta decyzja architektoniczna znaczaco upraszcza wdrazanie w srodowiskach kontenerowych, poniewaz programisci i analitycy danych moga uruchamiac zadania Spark bezposrednio z notebooków Jupyter lub potoków CI/CD, bez koniecznosci instalowania pelnej dystrybucji Spark na lokalnym komputerze.

# 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)

Protokól Spark Connect oparty jest na gRPC i serializuje logiczne plany zapytan zamiast przesylac dane miedzy klientem a serwerem. Prowadzi to do drastycznej redukcji narzutu sieciowego i umozliwia scenariusze multi-tenancy, w których setki uzytkowników jednoczesnie korzystaja z tego samego klastra. Dla przedsiebiorstw oznacza to lepsze wykorzystanie zasobów i uproszczone zarzadzanie dostepem, poniewaz autoryzacja odbywa sie przez znormalizowane punkty koncowe.

API transformWithState dla zlozonej logiki streamingowej

Nowe API transformWithState zastepuje przestarzale mapGroupsWithState i flatMapGroupsWithState bezpiecznym typowo, komponowalnym interfejsem. Kluczowa zaleta polega na jawnym zarzadzaniu stanem z TTL (Time-to-Live), co umozliwia automatyczne czyszczenie nieaktywnych sesji i tym samym rozwiazuje czesty problem w produkcyjnych potokach streamingowych.

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
      }
    }
  }
}

Konfiguracja TTL zapewnia, ze stan nieaktywnych sesji uzytkowników jest automatycznie usuwany po 30 minutach. Bez tego mechanizmu magazyn stanów w dlugo dzialajacych aplikacjach streamingowych rosólby w sposób ciagly, prowadzac ostatecznie do problemów z pamiecia. Na rozmowach kwalifikacyjnych czesto padaja pytania wlasnie o ten aspekt, poniewaz demonstruje on zrozumienie operacyjnych wyzwan systemów streamingowych.

Spark Declarative Pipelines (SDP) w Spark 4.1

Spark Declarative Pipelines (SDP) umozliwia definiowanie przepływów ETL w sposób deklaratywny. Zamiast recznego orkiestrowania planu wykonania, programista opisuje jedynie pozadane tabele i ich zaleznosci. Spark samodzielnie tworzy graf wykonania DAG, przeprowadza zrównoleglenie i obsluguje bledy.

# 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()

Koncepcja ta jest zblizona do dbt, jednak dziala bezposrednio w srodowisku uruchomieniowym Spark. Eliminuje to potrzebe zewnetrznych narzedzi orkiestracyjnych dla prostych i srednio zlozonych potoków oraz redukuje ogólna zlozonosc stosu Data Engineering. Szczególnie cenna jest automatyczna rozdzielczosc zaleznosci: gdy revenue_summary zalezy od daily_revenue, a to z kolei od clean_orders, Spark samodzielnie oblicza optymalna kolejnosc wykonania.

Pytania rekrutacyjne z Data Engineering dotyczace Spark 4

Ponizsze pytania pojawiaja sie regularnie na technicznych rozmowach kwalifikacyjnych na stanowiska Data Engineering i obejmuja kluczowe nowosci Spark 4.

Pytanie 1: Dlaczego Spark 4 wlaczyl tryb ANSI SQL jako domyslny i jakie ma to konsekwencje dla istniejacych potoków?

Tryb ANSI SQL poprawia jakosc danych poprzez rzucanie jawnych wyjatków zamiast cichego zwracania NULL przy nieprawidlowych konwersjach typów i przepelnieniach arytmetycznych. Istniejace potoki musza zamienic wszystkie wyrazenia CAST na TRY_CAST, aby zachowac dotychczasowe zachowanie. Zmiana ta ujawnia ukryte bledy danych, które wczesniej niezauwzone propagowaly sie przez caly potok przetwarzania.

Pytanie 2: Czym jest typ danych VARIANT i czym rózni sie od przetwarzania JSON opartego na STRING?

VARIANT przechowuje dane polstrukturalne w zoptymalizowanym formacie binarnym z automatycznym column shredding. W przeciwienstwie do kolumn STRING z from_json, VARIANT nie wymaga definicji schematu przy zapisie, a mimo to oferuje typowane dostepu przez notacje sciezkowa. Szybkosc zapytan jest nawet 10-krotnie wyzsza, poniewaz czesto odpytywane pola sa automatycznie materializowane jako osobne kolumny.

Pytanie 3: Czym rózni sie Real-Time Mode od dotychczasowego modelu Micro-Batch w Structured Streaming?

Model Micro-Batch przetwarza dane w dyskretnych przedzialach czasowych z typowymi opóznieniami od 100 ms do kilku sekund. Real-Time Mode przetwarza kazdy rekord natychmiast po jego nadejsciu, osiagajac opóznienia ponizej sekundy. Tryb ten jest odpowiedni dla aplikacji krytycznych czasowo, takich jak wykrywanie oszustw, wymaga jednak wiekszych zasobów i aktualnie wspiera jedynie scenariusze Kafka-do-Kafka.

Pytanie 4: Jakie zalety oferuje Spark Connect w porównaniu z tradycyjnym trybem klienckim?

Spark Connect calkowicie oddziela klienta od klastra za pomoca interfejsu gRPC. Klient nie potrzebuje lokalnej JVM, co upraszcza integracje w srodowiskach kontenerowych i notebookach. Logiczne plany zapytan sa serializowane i przesylane zamiast samych danych, co minimalizuje narzut sieciowy. Wielu uzytkowników moze jednoczesnie korzystac z tego samego klastra przez znormalizowane punkty koncowe.

Pytanie 5: Jaki problem rozwiazuje API transformWithState i dlaczego zostalo wprowadzone?

API transformWithState zastepuje mapGroupsWithState i flatMapGroupsWithState bezpiecznym typowo interfejsem z natywnym wsparciem TTL. Rozwiazuje problem nieograniczonego wzrostu stanu w dlugo dzialajacych aplikacjach streamingowych poprzez automatyczne czyszczenie nieaktywnych stanów. API umozliwia komponowalne definicje stanu z ValueState, ListState i MapState.

Kompletny potok ETL w Spark 4: Bronze-Silver-Gold

Ponizszy przyklad demonstruje gotowy do produkcji potok ETL, który integruje wszystkie nowe funkcje Spark 4: tryb ANSI SQL, typ danych VARIANT oraz Structured Streaming z Kafka jako zródlem i Delta Lake jako miejscem docelowym.

# 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()

Warstwa Bronze przyjmuje surowe dane z Kafki i konwertuje je za pomoca parse_json bezposrednio do typu danych VARIANT. W warstwie Silver nastepuje typowana ekstrakcja poszczególnych pól za pomoca notacji sciezkowej. Tryb ANSI zapewnia, ze bledne konwersje zostana natychmiast wykryte zamiast propagowac sie jako wartosci NULL do warstwy Gold. Warstwa Gold zapisuje oczyszczone i typowane dane do Delta Lake z wlaczona ewolucja schematu.

Ta architektura reprezentuje aktualnie najlepsza praktyke dla streamingowego ETL z uzyciem Spark 4 i laczy wszystkie istotne nowosci w spójny system.

Podsumowanie

Apache Spark 4 wyznacza zmiane generacyjna w przetwarzaniu Big Data. Wprowadzone nowosci adresuja konkretne slabosci poprzednich wersji i pozycjonuja Spark jako kompleksowa platforme zarówno dla przetwarzania wsadowego, jak i przetwarzania w czasie rzeczywistym.

Najwazniejsze usprawnienia w skrócie:

• Tryb ANSI SQL jako standard poprawia jakosc danych poprzez jawna obsluge bledów przy konwersjach typów
• Typ danych VARIANT upraszcza prace z danymi polstrukturalnymi przy jednoczesnym znacznym przyspieszeniu zapytan
• Real-Time Mode Streaming obniza opóznienie end-to-end do poziomu ponizej sekundy dla krytycznych czasowo zastoswan
• Spark Connect oddziela klienta od klastra i umozliwia lekkie integracje bez lokalnej JVM
• API transformWithState oferuje bezpieczne typowo zarzadzanie stanem z automatycznym czyszczeniem opartym na TTL
• Spark Declarative Pipelines redukuja zlozonosc przepływów ETL poprzez automatyczna rozdzielczosc zaleznosci

Dla inzynierów danych i programistów przygotowujacych sie do rozmów kwalifikacyjnych znajomosc tych funkcji jest obecnie niezbedna. Zdolnosc wyjasnienia róznic miedzy Micro-Batch a Real-Time Mode, zademonstrowania przewag VARIANT nad przetwarzaniem JSON opartym na STRING oraz zbudowania kompletnego potoku Bronze-Silver-Gold w Spark 4 stanowi fundament udanych rozmów rekrutacyjnych w obszarze Data Engineering.