Spark Streaming Notes

Apache Spark Streaming

Outline

• Streaming intro
• Basic concepts and terminology
• Sources and sinks for Structured Streaming
• Operations over Streaming Dataframe
• Micro-batch vs continuous mode
• Extras

Streaming Intro

• Real-time processing of continuous streams of data in motion.
• Major components:
  • Message processors: Deliver data for processing.
  • Stream processors: Processing layer which runs computations/application logic on the data.
  • Storage/Output: Stores results, prepares stream for other consumers, sends notifications, etc.

Batch Processing vs Stream Processing

Streaming in Spark

• Spark Streaming:
  • Legacy project, no longer updated.
  • DStreams – low level RDD streaming API.
• Spark Structured Streaming:
  • Streaming API built on the Spark SQL engine.
  • Optimizations of execution plans available.
  • Unified API for batch/streaming – code can be reused.
  • You can use DataSet/DataFrame API in Java, Scala, Python and R.
  • Internally, queries are processed in micro-batches.
  • Since Spark 2.3 – introduced continuous processing.

Basic Concept and Terminology

• Treat a live data stream as a table that is being continuously appended.
• Data stream as an unbounded table.

Example

Create stream and add logic:

lines = spark \
.readStream\
.format("socket") \
.option("host", "localhost")\
.option("port", 9999)\
.load()

words = lines.select(
    explode(
        split(lines.value, " ")
    ).alias("word")
)

wordCounts =
words.groupBy("word").count()

Start receiving data:

query = wordCounts \
    .writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

query.awaitTermination()

• Structured streaming does not materialize an entire table. Process latest data incrementally, update the result and discard. Keeps only minimal intermediate state that is required to update the result.

• Fault-tolerant.

  • Checkpoints
    • Metadata/data checkpoints saved to durable/fault-tolerant storage (S3, HDFS,..).
  • Write-ahead logs
    • Capture ingested data, but not yet processed by query.

• Handling recovery after system failure in streaming systems:

  • Fault-tolerance semantics:
    • At least once: Each message is guaranteed to be processed, but it may get processed more than once.
    • At most once: Each message may or may not be processed. If a message is processed, it's only processed once.
    • Exactly once: Each message is guaranteed to be processed once and only once.

How to Achieve Exactly-Once Delivery (Simplified)?

• Streaming source: In case of failure, data should be replayable in the source system.

• Checkpointing and write ahead logs: Store current state to durable, fault-tolerant storage to recover in case of driver/executor failures.

• Idempotent processing and sinks: Ensure that data is not duplicated when reprocessed/retried after failure. Ensure correct final state of the system after data is reprocessed.

• Output Modes:

  • Append: Only the new rows appended in the Result Table since the last trigger will be written to the external storage. This is applicable only on the queries where existing rows in the Result Table are not expected to change.
  • Complete: The entire updated Result Table will be written to the external storage. It is up to the storage connector to decide how to handle writing of the entire table.
  • Update: Only the rows that were updated in the Result Table since the last trigger will be written to the external storage (available since Spark 2.1.1).

Sources and Sinks for Structured Streaming

Data Sources for Structured Streaming

• File source
  • Reads file written in a directory as a stream of data
  • CSV, JSON, ORC, Parquet, S3,…
• Messaging services
  • Kafka, Kinesis, Event Hubs,…

df = (spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "<server: ip>")
.option("subscribe", "<topic>")
.option("startingOffsets", "latest")
.load()
)

val kinesis = spark.readStream
.format("kinesis")
.option("streamName", kinesisStreamName)
.option("region", kinesisRegion)
.option("initialPosition", "TRIM_HORIZON")
.option("awsAccessKey", awsAccessKeyId)
.option("aws SecretKey", awsSecretKey)
.load()

• Delta table
  • Incremental read of delta tables.
  • You can use change data feed (CDF) of Delta Lake table to upsert changes in downstream tables.
• Configure input rate
  • maxFilesPerTrigger: How many new files to be considered in every micro-batch (default 1000).
  • maxBytesPerTrigger: How much data gets processed in each micro-batch. This is not set by default.
• Ingesting supported files from cloud object storage
  • Classic file source or Databricks Auto Loader (cloudFiles format).
  • Auto Loader scales to support (near) real-time ingestion of millions of files per hour
  • S3, Azure Blob Storage, Google Cloud Storage,…
  • cloudFiles.schemaLocation option: supports schema inference and evolution.
• Benefits of Auto Loader over classic file source:
  • Efficient and more performant file discovery.
  • Schema inference and evolution.
  • Cheap file discovery
    • File detection modes
      • Directory listing
        • Used by default
        • Reduced number of API calls by listing files in subdirectories
      • File notification service
        • Leverages file notifications and queue service in cloud infrastructure account
        • Better for large input directories / high volume of files, more difficult to set up

Output Sinks for Structured Streaming

• File sink

resultDf
.writeStream
.outputMode("append") // Filesink only support Append mode.
.format("csv") // supports these formats: csv, json, orc, parquet
.option("path", "output/filesink_output")
.option("header", true)
.option("checkpointLocation", "checkpoint/filesink_checkpoint")
.start()
.awaitTermination()

• Kafka sink

(spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "<server:ip>")
.option("subscribe", "<topic>")
.option("starting Offsets", "latest")
.load()
.join(spark.read.table("<table-name>"), on="<id>", how="left")
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "<server:ip>")
.option("topic", "<topic>")
.option("checkpoint Location", "<checkpoint-path>")
.start()
)

• Delta Lake table sink
  • Good practice to ingest streaming data from external sources to Delta Lake.
  • Secured „exactly-once“ processing, enabled by transaction log.
• forEachBatch
  • Data are processed with custom logic per micro-batch -> can use (theoretically) arbitrary sink
  • Provides only at-least-once guarantees
  • You can use batchId to deduplicate data and secure exactly-once, but requires additional effort
  • Does not work with continuous-processing mode

Operations over Streaming Dataframe

• Most operations work just the same as in the case of classic dataframe
  • Select, where, groupBy,…
  • Temporary tables + SQL
  • Joins
    • Stream dataframe – Static dataframe
    • Stream dataframe – Stream dataframe
• We generally distinguish between two types of operations:
  • Stateless
    • E.g. Filter - needs only information available in current micro-batch
  • Stateful
    • Such as count of keys over 5 minute period (aggregation), drop duplicates, etc. - need to preserve state and get information about previous data/results
• Without restriction, state can become unbounded - will quickly introduce latency or even errors
• We should setup some threshold for how long to continue processing updates for a given state – watermark

Output Modes with Watermark

• Append with watermark: Rows are written to the target table once the watermark threshold has passed. Old state is dropped once the threshold has passed.
• Update with watermark: Rows are written to the target table as results are calculated, and can be updated and overwritten as new data arrives. Old state is dropped once the threshold has passed.
• Complete: Aggregation state is not dropped. The target table is rewritten with each trigger.

Window Operations

• Window operations over event-time

windowedCounts = words
    .groupBy(
        window(words.timestamp, "10 minutes", "5 minutes"),
        words.word
    )
    .count()

10 minute window, slide every 5 minutes

• Introduce watermark to handle late data
  • Old data arriving after watermark threshold has passed is not taken into account
  • Conditions to use watermark:
    • Update or append mode
    • Aggregation should use event_time based column or window function over event_time column
    • Watermark should be specified over same column as given aggregation
    • withWatermark clause must precede given aggregation

windowedCounts = words
    .withWatermark("timestamp", "10 minutes")
    .groupBy(
        window(words.timestamp, "10 minutes", "5 minutes"),
        words.word
    )
    .count()

Trigger

• Allow flexibility over time interval triggers – if you don‘t need real time processing, there is no need to have it – control cost
  • E.g. Update database every hour,…
    • .trigger(processingTime='10 seconds')
    • Default 500ms Micro-batch mode
  • .trigger(once=True)
    • Process all available data in single batch and exit, now deprecated
  • .trigger(availableNow=True)
    • Process all available data in multiple micro-batches and exit
    • Better scalability then „once“
  • .trigger(continuous= '1 second')
    • Low-latency, continuous mode

Micro-batch vs Continuous Mode

• Micro-batch Processing
  • Micro-batch Processing uses periodic tasks to process events
  • Second-scale end-to-end latencies with Micro-batch Processing
  • Spark driver launches short tasks in every micro-batch to process events.
  • To-be-processed offsets saved to a write-ahead-log before starting micro-batch
• Continuous processing
  • Continuous Processing uses long-running tasks to continuously process events.
  • Millisecond-scale end-to-end latencies with Continuous Processing
  • Driver launches long-running tasks at the start of the query
  • Tasks process events as soon as they are available at source.
  • Processed offsets saved to a write-ahead-log after every epoch.

Extras

Other Streaming Services

• Apache Flink
  • Real-time streaming
  • Also supports batch processing
• Apache Storm
  • Real-time streaming
• Kafka Streams