Apache Beam vs Spark 2026年版：統一パイプラインと面接質問の完全ガイド

# beam_pipeline.py
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# Same code runs on Dataflow, Flink, or Spark runner
options = PipelineOptions([
    '--runner=DataflowRunner',  # Switch to FlinkRunner or SparkRunner
    '--project=my-project',
    '--region=us-central1',
    '--temp_location=gs://my-bucket/temp'
])

with beam.Pipeline(options=options) as pipeline:
    (pipeline
     | 'ReadEvents' >> beam.io.ReadFromPubSub(topic='projects/p/topics/events')
     | 'ParseJSON' >> beam.Map(lambda x: json.loads(x))
     | 'FilterValid' >> beam.Filter(lambda e: e.get('status') == 'valid')
     | 'WindowByMinute' >> beam.WindowInto(beam.window.FixedWindows(60))
     | 'CountPerWindow' >> beam.combiners.Count.Globally()
     | 'WriteToBQ' >> beam.io.WriteToBigQuery('project:dataset.table'))

# spark_streaming.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window

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

# Spark 4.2: ANSI mode enabled by default, stricter type checking
events = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "broker:9092") \
    .option("subscribe", "events") \
    .load()

processed = events \
    .select(from_json(col("value").cast("string"), schema).alias("data")) \
    .filter(col("data.status") == "valid") \
    .groupBy(window(col("data.timestamp"), "1 minute")) \
    .count()

processed.writeStream \
    .format("bigquery") \
    .option("table", "project.dataset.table") \
    .outputMode("append") \
    .start()

# beam_sessions.py
from apache_beam import window

# 30-minute session gap, allow 1 hour late data
windowed = (
    events
    | 'SessionWindow' >> beam.WindowInto(
        window.Sessions(30 * 60),  # 30 min gap closes session
        trigger=beam.trigger.AfterWatermark(
            early=beam.trigger.AfterProcessingTime(60),
            late=beam.trigger.AfterCount(1)
        ),
        allowed_lateness=3600,  # Accept data up to 1 hour late
        accumulation_mode=beam.trigger.AccumulationMode.ACCUMULATING
    )
)

# spark_sessions.py
from pyspark.sql.streaming import GroupState, GroupStateTimeout

def update_session(key, events, state: GroupState):
    # Manual session management with state API
    session_data = state.getOption() or {"count": 0, "start": None, "end": None}
    
    for event in events:
        ts = event.timestamp
        if session_data["end"] and (ts - session_data["end"]).seconds > 1800:
            # Gap exceeded 30 min, emit previous session
            yield session_data
            session_data = {"count": 0, "start": ts, "end": ts}
        
        session_data["count"] += 1
        session_data["end"] = ts
        if not session_data["start"]:
            session_data["start"] = ts
    
    state.update(session_data)
    state.setTimeoutDuration(30 * 60 * 1000)  # 30 min timeout

# Spark 4.2 Arbitrary State API v2
result = events \
    .groupByKey(lambda e: e.user_id) \
    .flatMapGroupsWithState(
        update_session,
        outputMode="append",
        stateType=session_schema,
        timeoutConf=GroupStateTimeout.ProcessingTimeTimeout
    )

# Beam: exactly-once via runner guarantees
# Dataflow provides exactly-once for both sources and sinks
# The SDK handles deduplication and checkpoint coordination

with beam.Pipeline() as p:
    (p
     | beam.io.ReadFromPubSub(subscription='...')  # Exactly-once read
     | beam.Map(process)
     | beam.io.WriteToBigQuery(...)  # Exactly-once write with retries
    )

# Spark: exactly-once via checkpointing and idempotent sinks
spark.readStream \
    .format("kafka") \
    .load() \
    .writeStream \
    .option("checkpointLocation", "/checkpoint")  # State recovery
    .foreachBatch(idempotent_write)  # Application-level deduplication
    .start()

[ストリーミング取り込み]     [バッチ処理]     [ML訓練]
        |                        |                    |
   Beam/Dataflow           Spark on Databricks    Spark MLlib
        |                        |                    |
        v                        v                    v
    BigQuery  <----- dbt -----> Delta Lake -----> Model Registry