Apache Spark 4: ฟีเจอร์ใหม่ Structured Streaming และคำถามสัมภาษณ์งาน

Apache Spark 4 เป็นเวอร์ชันที่มีการเปลี่ยนแปลงครั้งใหญ่ที่สุดนับตั้งแต่ Spark 3.0 เปิดตัวในปี 2020 การอัปเดตครั้งนี้นำเสนอความสามารถใหม่ที่สำคัญสำหรับ data engineers และ platform teams ที่ต้องการสร้างระบบ data processing ที่ทันสมัยและมีประสิทธิภาพสูง ฟีเจอร์เด่นที่โดดเด่นที่สุดคือการเปิดใช้งาน ANSI SQL mode โดยค่าเริ่มต้น การเพิ่ม VARIANT data type สำหรับจัดการข้อมูล semi-structured และ Real-Time Mode ใน Structured Streaming ที่ลด latency ลงเหลือระดับ sub-second

การเปลี่ยนแปลงเหล่านี้ไม่ได้เป็นเพียงการปรับปรุงเล็กน้อย แต่เป็นการออกแบบใหม่ที่ส่งผลต่อวิธีการเขียนโค้ด การจัดการ schema และการตอบสนองต่อข้อมูลแบบ real-time องค์กรที่ใช้ Spark ในระบบ production จำเป็นต้องเข้าใจการเปลี่ยนแปลงเหล่านี้อย่างถ่องแท้ เนื่องจากมีผลกระทบโดยตรงต่อ backward compatibility และ performance characteristics ของ data pipelines ที่มีอยู่

ANSI SQL Mode: การเปลี่ยนแปลงที่สำคัญที่สุด

การเปิดใช้งาน ANSI SQL mode โดยค่าเริ่มต้นเป็นการเปลี่ยนแปลงที่มีผลกระทบมากที่สุดใน Spark 4 ใน Spark 3.x พฤติกรรมเริ่มต้นจะให้ค่า NULL เมื่อเกิด type casting ที่ล้มเหลวหรือการหารด้วยศูนย์ แต่ใน ANSI mode ของ Spark 4 การดำเนินการเหล่านี้จะ throw exception ทันทีแทน ทำให้ระบบมีความเข้มงวดและปลอดภัยมากขึ้น แต่ต้องปรับปรุงโค้ดเดิมเพื่อรองรับ

การเปลี่ยนแปลงนี้มีจุดประสงค์เพื่อให้ Spark ทำงานสอดคล้องกับ SQL standard มากขึ้นและป้องกันข้อผิดพลาดที่เกิดจากการประมวลผลข้อมูลที่ไม่ถูกต้องโดยไม่รู้ตัว ในระบบ production ที่มีข้อมูลหลายล้านหรือหลายพันล้านแถว การที่มี silent NULL values อาจทำให้เกิดผลลัพธ์ที่ผิดพลาดโดยไม่มีการแจ้งเตือน ANSI mode บังคับให้ developer จัดการกับ data quality issues อย่างชัดเจน

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

ตัวอย่างโค้ดด้านบนแสดงการใช้ TRY_CAST แทน CAST ปกติเพื่อรักษาพฤติกรรมการคืนค่า NULL เมื่อการแปลงล้มเหลว นี่เป็นกลยุทธ์การ migrate ที่ปลอดภัยที่สุดสำหรับ data pipelines ที่มีอยู่ นอกจากนี้ยังมี functions อื่นๆ เช่น TRY_DIVIDE, TRY_ADD, TRY_MULTIPLY ที่สามารถใช้เพื่อหลีกเลี่ยง arithmetic exceptions

VARIANT Data Type: จัดการ Semi-Structured Data อย่างมีประสิทธิภาพ

VARIANT data type เป็นนวัตกรรมสำคัญที่ช่วยให้ Spark จัดการกับข้อมูล JSON และ semi-structured data ได้อย่างมีประสิทธิภาพมากขึ้น ก่อนหน้านี้ การทำงานกับ nested JSON ใน Spark ต้องใช้ from_json() และกำหนด schema อย่างชัดเจน หรือใช้ schema_of_json() ซึ่งมีค่าใช้จ่ายในการประมวลผลสูง VARIANT type ช่วยให้เก็บข้อมูล JSON ในรูปแบบ native และ query ได้โดยตรงด้วย dot notation

ภายในระบบ VARIANT ใช้เทคนิค shredding เพื่อแยกฟิลด์ที่ใช้บ่อยออกมาเก็บแยกต่างหาก ทำให้การ query เร็วขึ้นอย่างมาก โดยเฉพาะกับ columnar storage formats เช่น Parquet การใช้ VARIANT ช่วยลดความซับซ้อนของโค้ดและเพิ่มความยืดหยุ่นในการจัดการกับข้อมูลที่มี schema ไม่แน่นอนหรือเปลี่ยนแปลงบ่อย

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

ตัวอย่างนี้แสดงการสร้างตารางที่มี column ประเภท VARIANT และการ query ข้อมูล nested fields ด้วย syntax payload:user_id::INT ที่อ่านง่ายและมีประสิทธิภาพสูง การใช้ PARSE_JSON() function จะแปลง JSON string เป็น VARIANT type โดยอัตโนมัติ ซึ่งเหมาะสำหรับ use cases เช่น event logging, user analytics และ IoT data ingestion

Real-Time Mode: Structured Streaming ความเร็วระดับ Sub-Second

Real-Time Mode เป็นการปรับปรุง Structured Streaming engine ให้มี latency ต่ำลงอย่างมากจาก micro-batch processing ปกติที่มี latency อยู่ที่ 1-5 วินาที Real-Time Mode ลดลงเหลือ 100-500 milliseconds การปรับปรุงนี้ทำได้ด้วยการออกแบบ scheduling และ checkpointing ใหม่ที่ไม่ต้องรอให้ micro-batch เสร็จสมบูรณ์ก่อนเริ่ม batch ถัดไป

สิ่งนี้มีความสำคัญอย่างยิ่งสำหรับ use cases ที่ต้องการความเร็วสูง เช่น fraud detection, real-time recommendations และ operational monitoring ที่ความล่าช้าแม้เพียงไม่กี่วินาทีอาจส่งผลกระทบต่อธุรกิจอย่างมีนัยสำคัญ อย่างไรก็ตาม Real-Time Mode มีข้อแลกเปลี่ยน คือใช้ resources มากขึ้นและต้องการ tuning ที่ละเอียดกว่า

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

ตัวอย่างนี้แสดง streaming pipeline ที่อ่านข้อมูลจาก Kafka กรองค่า temperature ที่เกิน threshold และส่งต่อไปยัง alert topic การตั้งค่า spark.sql.streaming.mode เป็น realtime จะเปิดใช้งาน Real-Time Mode ซึ่งเหมาะสำหรับ scenarios ที่ต้องการ latency ต่ำที่สุด

Spark Connect: Thin Client Architecture

Spark Connect เป็น client-server architecture ใหม่ที่แยก client process ออกจาก Spark driver ใน Spark 3.x client ต้องรัน JVM และโหลด Spark dependencies ทั้งหมด ซึ่งทำให้ memory footprint สูงและ startup time ช้า Spark Connect ใช้ thin client ที่สื่อสารกับ Spark cluster ผ่าน gRPC protocol ทำให้ client เบาและเริ่มต้นเร็วกว่ามาก

สถาปัตยกรรมนี้เปิดโอกาสให้ใช้ Spark จาก environments ที่มีข้อจำกัด เช่น serverless functions, edge devices หรือ notebooks ที่มี resources จำกัด นอกจากนี้ยังช่วยเพิ่มความปลอดภัยเนื่องจาก client ไม่ต้องมี direct access ไปยัง data sources และสามารถควบคุม permissions ได้ที่ server side

# 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 cluster แบบ remote โดยใช้ remote() method แทน master() ปกติ ทุกการดำเนินการ DataFrame จะถูกส่งไปประมวลผลที่ server ส่วน client แค่รับผลลัพธ์กลับมา ทำให้สามารถ scale processing capacity ได้อย่างอิสระจาก client resources

transformWithState API: Stateful Streaming ที่ทรงพลัง

transformWithState เป็น API ใหม่สำหรับจัดการ stateful operations ใน Structured Streaming ที่มีความยืดหยุ่นมากกว่า mapGroupsWithState และ flatMapGroupsWithState ที่มีอยู่ใน Spark 3.x ฟีเจอร์เด่นที่สำคัญคือการรองรับ TTL (Time-To-Live) สำหรับ state ทำให้สามารถทำ automatic cleanup ของ state ที่ไม่ได้ใช้งานได้โดยไม่ต้องเขียน logic เอง

การมี TTL support นี้แก้ปัญหาหลักของ stateful streaming คือการเติบโตของ state อย่างไม่มีที่สิ้นสุด ในระบบ production หลายครั้งที่ state size เพิ่มขึ้นเรื่อยๆ จนทำให้ checkpointing ช้าและใช้ storage มาก transformWithState ช่วยจัดการปัญหานี้โดยอัตโนมัติ

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

ตัวอย่างนี้แสดง session tracking ที่จับเวลาและนับจำนวน actions ต่อ user โดยใช้ ValueState พร้อม TTL 30 นาที หมายความว่าถ้า user ไม่มี activity ภายใน 30 นาที state จะถูกลบอัตโนมัติ ซึ่งเหมาะสำหรับ use cases เช่น user session analytics, fraud detection และ real-time personalization

Spark Declarative Pipelines: Data Engineering แบบ Functional

Spark Declarative Pipelines (SDP) เป็น framework สำหรับเขียน data pipelines แบบ declarative โดยกำหนดเพียงว่าต้องการ output อะไร แทนที่จะระบุขั้นตอนการประมวลผลทีละขั้นตอน Spark จะสร้าง execution plan และจัดการ dependencies ระหว่าง datasets เอง รวมถึง error handling และ retries อย่างอัตโนมัติ

แนวคิดนี้ใกล้เคียงกับ dbt (data build tool) ในโลก SQL-first data engineering แต่มีความสามารถเต็มรูปแบบของ Spark รองรับทั้ง batch และ streaming และสามารถใช้ custom transformations ที่ซับซ้อนได้ SDP ช่วยลดความซับซ้อนของ orchestration และทำให้ pipelines maintainable มากขึ้น

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

โค้ดนี้แสดง pipeline สามขั้นตอนที่ Spark จะวิเคราะห์ dependencies และ execute ในลำดับที่เหมาะสม หาก daily_revenue ล้มเหลว Spark จะ retry อัตโนมัติโดยไม่กระทบกับ tables อื่นที่ไม่เกี่ยวข้อง การใช้ decorators ทำให้โค้ดอ่านง่ายและ testable

คำถามสัมภาษณ์งาน Apache Spark 4

1. อธิบายความแตกต่างระหว่าง ANSI mode และ non-ANSI mode ใน Spark 4

ANSI mode ช่วยเพิ่ม data quality และป้องกัน silent failures ที่อาจทำให้ business logic ผิดพลาดโดยไม่รู้ตัว อย่างไรก็ตาม การ migrate จาก Spark 3.x ต้องทดสอบอย่างระมัดระวัง เพราะโค้ดที่เคยทำงานได้อาจเกิด exceptions ใน Spark 4 การใช้ TRY_CAST, TRY_DIVIDE และ functions ใน try family อื่นๆ เป็นวิธี migrate ที่ปลอดภัย

2. VARIANT data type แตกต่างจาก STRING column ที่เก็บ JSON อย่างไร

VARIANT ยังรองรับ indexing และ statistics collection ทำให้ Catalyst optimizer สามารถสร้าง efficient query plans ได้ นอกจากนี้ยังช่วยลดขนาดข้อมูลเมื่อใช้กับ Parquet หรือ Delta Lake เพราะ columnar compression ทำงานได้ดีกว่ากับ shredded data

3. Real-Time Mode ใน Structured Streaming ใช้ในสถานการณ์ใดบ้าง

ข้อแลกเปลี่ยนของ Real-Time Mode คือ throughput อาจลดลงเล็กน้อยเมื่อเทียบกับ micro-batch mode ปกติ เนื่องจาก overhead ของ frequent checkpointing และ coordination นอกจากนี้ต้องตั้งค่า spark.sql.streaming.minBatchesToRetain และ checkpoint intervals อย่างเหมาะสมเพื่อสมดุลระหว่าง latency กับ recovery time

4. อธิบาย transformWithState API และข้อดีเหนือ mapGroupsWithState

การมี TTL support หมายความว่าไม่ต้องเขียน logic เพื่อลบ state ที่เก่าเกินไปเอง ซึ่งช่วยลดความซับซ้อนของโค้ดและป้องกัน memory leaks นอกจากนี้ transformWithState ยังรองรับ processing time และ event time timers ทำให้สามารถสร้าง complex stateful logic เช่น session windows, pattern detection และ temporal joins ได้ง่ายขึ้น

5. Spark Connect architecture มีประโยชน์อย่างไรใน production

Spark Connect ยังช่วยในการทำ multi-tenancy เพราะสามารถควบคุม authentication, authorization และ resource limits ได้ที่ server side การใช้ gRPC protocol ทำให้การสื่อสารมีประสิทธิภาพและรองรับ streaming results ได้ดีกว่า REST API ทั่วไป

ETL Pipeline สมบูรณ์ด้วย Spark 4

ตัวอย่างนี้แสดง end-to-end ETL pipeline ที่ใช้ฟีเจอร์หลักของ Spark 4 รวมกัน ประกอบด้วย bronze layer สำหรับ raw ingestion, silver layer สำหรับ data cleaning และ type extraction และ gold layer สำหรับ analytics-ready data โครงสร้างแบบ medallion architecture นี้เป็น best practice สำหรับ modern data platforms

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

Pipeline นี้ใช้ VARIANT type ใน bronze layer เพื่อเก็บ JSON payload โดยไม่ต้องกำหนด schema ล่วงหน้า จากนั้นใน silver layer จึงแยกฟิลด์ที่ต้องการออกมาพร้อม type casting ที่ชัดเจน การเปิด ANSI mode ทำให้ data quality checks เข้มงวดขึ้น และการใช้ Delta Lake ใน gold layer ช่วยให้ได้ ACID guarantees พร้อม time travel capabilities

สรุป

Apache Spark 4 นำเสนอการปรับปรุงที่สำคัญหลายประการที่ส่งผลต่อวิธีการสร้างและ operate data platforms ในยุคใหม่ ฟีเจอร์หลักที่ควรให้ความสำคัญมากที่สุดคือ

• ANSI SQL Mode: เพิ่ม data quality และความปลอดภัยโดย throw exceptions แทนการคืนค่า NULL อย่างเงียบๆ ต้องวางแผน migration อย่างระมัดระวัง
• VARIANT Data Type: ปฏิวัติการจัดการข้อมูล semi-structured ด้วย native support และ query performance ที่เหนือกว่า
• Real-Time Mode: ลด streaming latency ลงเหลือ sub-second เหมาะสำหรับ use cases ที่ต้องการความเร็วสูง
• Spark Connect: เปิดโอกาสใช้งาน Spark จาก environments ที่มีข้อจำกัดและเพิ่มความปลอดภัย
• transformWithState: ทำให้ stateful streaming สร้างและ maintain ง่ายขึ้นด้วย TTL support
• Declarative Pipelines: ลดความซับซ้อนของ orchestration ด้วยแนวคิด functional programming

การอัปเกรดจาก Spark 3.x ไปยัง Spark 4 ต้องทดสอบอย่างละเอียด โดยเฉพาะส่วนที่เกี่ยวกับ ANSI mode และ deprecated APIs องค์กรควรเริ่มจากการรัน test suites กับ Spark 4 ใน staging environment ก่อนนำไปใช้ใน production อย่างไรก็ตาม ประโยชน์ในด้าน performance, data quality และความยืดหยุ่นที่ได้รับคุ้มค่ากับความพยายามในการ migrate อย่างแน่นอน