Apache Spark 4: Yeni Ozellikler, Structured Streaming ve Mulakat Sorulari (2026)

Apache Spark 4, versiyon 3.0'dan bu yana gerceklestirilen en kapsamli guncellemeyi temsil etmekte ve Data Engineering ekosisteminin tumunu derinden etkilemektedir. ANSI SQL modunun varsayilan standart olarak etkinlestirilmesi, yari yapilandirilmis veriler icin tasarlanan VARIANT veri tipi ve tamamen yeniden yapilandirilan Streaming cercevesi ile Spark 4, modern veri pipeline'lari icin merkezi bir platform konumuna yukselmistir. Bu makale, Spark 4'un onemli yeniliklerini analiz etmekte, somut kod ornekleriyle pratik uygulamalarini gostermekte ve Data Engineering alanindaki teknik mulakat sorularina sistematik bir hazirlik sunmaktadir.

ANSI SQL Modu: Varsayilan Yeni Standart

Spark 4'teki en genis kapsamli degisikliklerden biri, ANSI SQL modunun varsayilan yapilandirma olarak etkinlestirilmesidir. Onceki surumlerde hatali tip donusumleri sessizce NULL degeri dondururken, Spark 4 artik ArithmeticException firlatmaktadir. Bu davranis degisikligi veri kalitesini onemli olcude artirmakta, ancak mevcut pipeline'larin sistematik bir sekilde goc ettirilmesini gerektirmektedir.

Gecis surecinde en guvenilir yontem, onceki NULL davranisini koruyarak ANSI uyumlu calisan TRY_CAST fonksiyonunun kullanilmasidir.

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

Buyuk kod tabanlarini goc ettiren ekiplerin, ilk adim olarak tum CAST ifadelerini tespit edip TRY_CAST ile degistirdikleri kapsamli bir denetim asamasi planlamasi onerilmektedir. Pratikte en iyi yaklasim, ANSI modunu once staging ortamlarinda etkinlestirmek ve ortaya cikan exception'lari sistematik olarak kayit altina almaktir. Bu sekilde, degisiklik uretime aktarilmadan once tum sorunlu noktalar belirlenmis olur.

VARIANT Veri Tipi: Yari Yapilandirilmis Veriler Icin Cozum

Yeni VARIANT veri tipi, Spark dnyasinda uzun suredir var olan bir sorunu cozmektedir: JSON, Avro veya Protobuf gibi yari yapilandirilmis verilerin verimli islenmesi. Onceki surumlerde ic ice gecmis JSON yapilariyla calismak, ya zahmetli sema tanimlamalari ya da String parsing gecici cozumleri gerektiriyordu. VARIANT, verileri otomatik shredding ile optimize edilmis bir ikili formatta saklamakta ve ic ice gecmis alanlara erisimi geleneksel JSON parsing'e kiyasla belirgin sekilde hizlandirmaktadir.

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'in klasik STRING sutunlari ve from_json yaklasimina gore temel avantaji, depolama verimliligi ve sorgu hizinda yatmaktadir. Otomatik shredding mekanizmasi sayesinde Spark, sik sorgulanan alanlari ayri sutunlara cikarirken, daha az ihtiyac duyulan verileri sikistirilmis olarak saklamaktadir. Benchmark testlerinde VARIANT, ic ice gecmis JSON verileri uzerinde tipik sorgularda String tabanli islemeye kiyasla 10 kata kadar hizlanma saglamaktadir.

Data Engineer'lar icin bu, Kafka, REST API'ler veya IoT cihazlarindan gelen olay verilerinin onceden bir sema tanimlamaya gerek kalmadan dogrudan Spark tablolarina yazilabilmesi ve ardindan SQL sozdizimi ile sorgulanabilmesi anlamina gelmektedir.

Real-Time Mode Streaming: Saniye Alti Gecikme

Spark 4.1, isleme gecikmesini bir saniyenin altina dusuren Real-Time Mode'u yeni bir streaming secenegi olarak sunmaktadir. Mevcut Micro-Batch modeli verileri ayrik zaman araliklerinda islerken, Real-Time Mode surekli olarak calisir ve sonuclari aninda alt sistemlere iletir.

Yapilandirma, spark.sql.streaming.mode parametresinin realtime olarak ayarlanmasiyla gerceklestirilir.

# 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 ozellikle dolandiricilik tespiti, anlik uyari sistemleri ve operasyonel panolar gibi her saniyenin is acisinan kritik oldugu kullanim senaryolari icin uygundur. Ancak bu modun geleneksel Micro-Batch modeline kiyasla daha yuksek kaynak gereksinimleri sundugu unutulmamalidir. Dolayisiyla yalnizca saniye alti gecikmenin gercekten gerekli oldugu senaryolarda tercih edilmelidir.

Spark Connect: Hafif PySpark Istemcisi

Spark Connect, istemciyi kume ortamindan tamamen ayirmakta ve Spark islemlerinin yerel JVM gerektirmeyen ince bir istemci kutuphanesi uzerinden yurutulmesini mumkun kilmaktadir. Bu mimari karar, konteyner tabanli ortamlarda dagitimi onemli olcude basitlestirmektedir. Gelistiriciler ve veri bilimciler, yerel makinelerinde tam bir Spark kurulumu olmadan Jupyter Notebook'lar veya CI/CD pipeline'lari uzerinden dogrudan Spark isleri calistirabilmektedir.

# 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 protokolu gRPC uzerine insa edilmistir ve istemci ile sunucu arasinda veri tasimak yerine mantiksal sorgu planlarini serilebtirmektedir. Bu yaklasim, ag ek yukunu onemli olcude azaltmakta ve yuzlerce kullanicinin ayni kumeye es zamanli erisebilecegi coklu kiralama senaryolarini mumkun kilmaktadir. Kurumlar icin bu, daha iyi kaynak kullanimi ve standartlastirilmis erisim noktalari uzerinden yonetilen basitlestirilmis yonetisim anlamina gelmektedir.

transformWithState API: Karmasik Streaming Mantigi

Yeni transformWithState API'si, kullanim disi kalan mapGroupsWithState ve flatMapGroupsWithState yontemlerini tip guvenli ve birlestirilebilir bir arabirimle degistirmektedir. Temel avantaj, TTL (Time-to-Live) destekli acik durum yonetiminde yatmaktadir. Bu ozellik, aktif olmayan oturumlarin otomatik temizlenmesini saglayarak uretim streaming pipeline'larinda sikca karsilasilan bir sorunu cozmektedir.

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 yapilandirmasi, aktif olmayan kullanici oturumlarinin durumunun 30 dakika sonra otomatik olarak temizlenmesini saglamaktadir. Bu mekanizma olmadan, uzun sureli calisan streaming uygulamalarinda State Store surekli buyuyerek sonunda bellek sorunlarina yol acmaktadir. Teknik mulakatlarda tam olarak bu konuda soru sorulmaktadir; cunku bu, streaming sistemlerinin operasyonel zorluklarina dair derinlemesine bir anlayis gosterir.

Spark Declarative Pipelines (SDP)

Spark Declarative Pipelines (SDP), ETL is akislarinin bildirimsel bir sekilde tanimlanmasini mumkun kilmaktadir. Gelistirici, yurutme planini elle organize etmek yerine yalnizca istenen tablolari ve bunlarin bagimlilik iliskilerini tanimlamaktadir. Spark, yurutme DAG'inin olusturulmasini, paralellestirmeyi ve hata yonetimini otomatik olarak ustlenmektedir.

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

Bu yaklasim kavramsal olarak dbt'ye benzemekte, ancak dogrudan Spark calisma ortami icerisinde calismaktadir. Bu sayede basit ve orta karmasikliktaki pipeline'lar icin harici orkestrasyon araclarinin gerekliligi ortadan kalkmakta ve Data Engineering yigitinin genel karmasikligi azaltilmaktadir. Ozellikle otomatik bagimlilik cozumleme oldukca degerlidir: revenue_summary tablosu daily_revenue'ye, o da clean_orders'a bagimli oldugunda, Spark optimal yurutme sirasini bagimsiz olarak hesaplamaktadir.

Data Engineering Mulakat Sorulari: Spark 4

Asagidaki sorular, Data Engineering pozisyonlari icin yapilan teknik mulakatlarda duzenli olarak karsimiza cikmakta ve Spark 4'un temel yeniliklerini kapsamaktadir.

Soru 1: Spark 4 neden ANSI SQL modunu varsayilan olarak etkinlestirdi ve bunun mevcut pipeline'lar uzerindeki etkisi nedir?

ANSI SQL modu, gecersiz tip donusumlerini ve aritmetik tasmalari artik sessizce NULL olarak degerlendirmek yerine acik exception'lar firlatarak veri kalitesini artirmaktadir. Mevcut pipeline'larin, onceki davranisi korumak icin tum CAST ifadelerini TRY_CAST ile degistirmesi gerekmektedir. Bu degisiklik, daha once fark edilmeden pipeline boyunca yayilan gizli veri hatalarini gozler onune sermektedir.

Soru 2: VARIANT veri tipi nedir ve STRING tabanli JSON isleme yonteminden nasil farklilasir?

VARIANT, yari yapilandirilmis verileri otomatik sutun shredding ile optimize edilmis bir ikili formatta saklamaktadir. STRING sutunlari ve from_json yaklasiminin aksine, VARIANT depolama sirasinda sema tanimi gerektirmez ancak yol notasyonu uzerinden tiplandirilmis erisimler sunmaktadir. Sik sorgulanan alanlar otomatik olarak ayri sutunlar halinde materyalize edildigi icin sorgu hizi 10 kata kadar artmaktadir.

Soru 3: Real-Time Mode, Structured Streaming'deki mevcut Micro-Batch yonteminden nasil ayrilir?

Micro-Batch modeli verileri ayrik zaman araliklerinda isler ve tipik gecikme suresi 100ms ile birkac saniye arasindadir. Real-Time Mode ise her kaydi varis aninda aninda isleyerek saniye alti gecikme surelerine ulasir. Bu mod, dolandiricilik tespiti gibi zaman acisinan kritik uygulamalar icin uygundur, ancak daha fazla kaynak gerektirir ve su anda yalnizca Kafka-Kafka senaryolarini desteklemektedir.

Soru 4: Spark Connect, geleneksel istemci moduna kiyasla hangi avantajlari sunmaktadir?

Spark Connect, istemciyi gRPC arabirimi uzerinden kumeden tamamen ayirmaktadir. Istemci yerel JVM gerektirmez ve bu da konteyner ortamlari ve notebook'lara entegrasyonu basitlestirmektedir. Mantiksal sorgu planlari serilestirilerek aktarilir, verinin kendisi degil; bu da ag ek yukunu minimize eder. Birden fazla kullanici, standartlastirilmis erisim noktalari uzerinden ayni kumeyi kullanabilmektedir.

Soru 5: transformWithState API hangi sorunu cozmekte ve neden tanitilmistir?

transformWithState API, mapGroupsWithState ve flatMapGroupsWithState yontemlerini tip guvenli bir arabirim ve yerel TTL destegi ile degistirmektedir. Uzun sureli calisan streaming uygulamalarinda aktif olmayan durumlarin otomatik temizlenmesi yoluyla sinirsiz durum buyumesi sorununu cozmektedir. API, ValueState, ListState ve MapState ile birlestirilebilir durum tanimlamalarina olanak tanimaktadir.

Eksiksiz Spark 4 ETL Pipeline: Bronze-Silver-Gold

Asagidaki ornek, Spark 4'un tum yeni ozelliklerini entegre eden uretim kalitesinde bir ETL pipeline'ini gostermektedir: ANSI SQL modu, VARIANT veri tipi ve kaynak olarak Kafka, hedef olarak Delta Lake kullanan Structured Streaming.

# 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 katmani, ham verileri Kafka'dan alarak parse_json ile dogrudan VARIANT veri tipine donusturmektedir. Silver katmaninda, bireysel alanlarin tiplandirilmis cikarimi yol notasyonu uzerinden gerceklestirilir. ANSI modu, hatali donusumlerin Gold katmanina NULL degerleri olarak ulasmasi yerine aninda tespit edilmesini saglamaktadir. Gold katmani, temizlenmis ve tiplandirilmis verileri sema evrimi aktif edilmis sekilde Delta Lake'e yazmaktadir.

Bu mimari, Spark 4 ile streaming ETL icin guncel en iyi uygulama yaklasimini temsil etmekte ve tum temel yenilikleri tutarli bir sistemde bir araya getirmektedir.

Sonuc

Apache Spark 4, buyuk veri islemede bir nesil degisimini isaret etmektedir. Yenilikler, onceki surumlerin somut zayif noktalarini ele almakta ve Spark'i hem toplu islem hem de gercek zamanli veri isleme icin kapsamli bir platform olarak konumlandirmaktadir.

Temel gelismelerin ozeti:

• Varsayilan ANSI SQL modu, tip donusumlerinde acik hata yonetimi ile veri kalitesini artirmaktadir
• VARIANT veri tipi, yari yapilandirilmis verilerle calismayi basitlestirirken ayni zamanda belirgin sekilde daha yuksek sorgu hizi saglamaktadir
• Real-Time Mode Streaming, zaman acisinan kritik kullanim senaryolari icin uctan uca gecikmeyi saniye alti seviyesine dusurmektedir
• Spark Connect, istemci ile kumeyi ayirarak yerel JVM gerektirmeyen hafif entegrasyonlari mumkun kilmaktadir
• transformWithState API, otomatik TTL tabanli temizleme ile tip guvenli durum yonetimi sunmaktadir
• Spark Declarative Pipelines, otomatik bagimlilik cozumleme ile ETL is akislarinin karmasikligini azaltmaktadir

Data Engineer'lar ve teknik mulakatlara hazirlanan gelistiriciler icin bu ozelliklere hakimiyet artik vazgecilmez bir niteliktir. Micro-Batch ile Real-Time Mode arasindaki farklari aciklayabilmek, VARIANT'in STRING tabanli JSON parsing'e gore avantajlarini gosterebilmek ve Spark 4 ile eksiksiz bir Bronze-Silver-Gold pipeline olusturabilmek, Data Engineering alanindaki basarili mulakatlarin temelini olusturmaktadir.