Stream processing and Spark Streaming

1. From batch to stream: why streaming matters

Everything in Chapters 3 to 9 was batch analysis: complex computations over large amounts of stored data, whose results arrive after seconds, minutes or hours. Big data is not just that. It is also about analyzing data streams: simpler computations over smaller sets of continuously incoming data, with results produced near real-time.

Streaming data processing is a big deal in big data for two business reasons:

• Latency. Businesses crave ever more timely data. Switching from batch to streaming is a good way to achieve lower latency.
• Workload balancing. Processing data as it arrives spreads workloads out more evenly over time, yielding more consistent and predictable consumption of resources.

Despite this business-driven surge of interest, batch systems are generally more mature than their streaming brethren — which has resulted in a lot of active development in recent times. Keep that asymmetry in mind: most of the theory in this chapter is old (the algorithms date to the 1970s–2000s), while the engineering (the tools in sections 15 to 19) is still being stabilized.

Sources and use cases

2. Stream models and real-time classification

The lecture fixes a notation before anything else. Consider:

• S = […, i_t−1, i_t, i_t+1, …] — the long sequence of items arriving rapidly in the stream;
• E = { e₁, e₂, …, e_n } — the set of elements described by the items;
• elem(i_t) = e_m — the element described by a stream item.

What an item says about its element defines the data stream model. The three classical models differ in the relationship between consecutive items of the same element:

Separately, real-time systems are classified by their tolerance for delays. The three categories come from the embedded-systems vocabulary, with latencies measured on very different scales:

3. The characteristics of streaming systems

Three properties separate streaming from everything the previous chapters did. They are the constraints every design in this chapter must obey, so they are worth stating as a checklist:

• Infinite dataset. Data is always being generated, and there is no control over the order in which data elements arrive.
• Infinite computation. The system must be always on and must be able to keep up with the data. There must be a plan to avoid overflowing — e.g., auto-scalability.
• Low-latency, approximate and/or speculative results. Data can usually be processed a single time (one pass); only a fraction of the dataset can be kept in memory for analyses; approximation may be required to accommodate the low-latency requirement.

4. The streaming architecture

A streaming system is a pipeline of tiers, from the data producer to the data consumer. The lecture names six components plus two storage tiers:

Each tier has a single responsibility:

• Collection — receives and collects data.
• Message queueing — handles the exchange of data items between tiers (most importantly, from collection to analysis).
• Analysis — runs algorithms on the data.
• Data access — makes the data available to consumers: dashboards, APIs, and so on.
• In-memory storage — supports the analysis tier.
• Long-term storage — keeps the data for future batch analysis.

The collection and access components are typically made of edge servers that either receive data from external sources or expose the analyzed data to the consumer — and, possibly, the producers and consumers are other streaming pipelines, which is how streaming systems compose into larger ones. The collection tier has three characteristics worth knowing:

• Interaction patterns may differ: request/response, publish/subscribe, streaming.
• Fault-tolerance strategies differ too: e.g., sender/receiver based message logging.
• Formats: mostly JSON, but also Thrift, Protocol Buffers, Apache Avro, …

5. Message queueing: producer, broker, consumer

The message queueing tier handles the transportation and exchange of data between tiers. Why dedicate an entire tier to moving data? Three reasons, all from the lecture:

• Decoupling. By decoupling the pipeline of operations (collection, analysis, data access), each node in the cluster will do one job only.
• Safe communication. Message queueing provides a solid framework for a safe communication between such nodes — if the analysis tier fails, data would get lost without a queue to buffer it.
• Funneling. Message queueing handles the funneling of n data streams to m consumers: the collection tier just sends data to the queue, independently of how many analysis tiers want to access that data.

A bit of history

The lecture compresses the evolution of message queueing into three generations, and the progression is the chapter’s thesis in miniature:

Producer–Broker–Consumer

The core concepts are the producer and consumer of data within the architecture, and the broker, which manages one or more queues of data. The broker is usually distributed on multiple nodes. Queues are identified by a topic (i.e., a type of message) and usually partitioned; when consumers are ready, they check the queues for new data.

Topics are the naming device that keeps unrelated traffic apart, and the lecture lists three uses for them:

• distinguish different sources — Sensor-A, Sensor-B, …;
• distinguish different types of data items — Order, Payments, Inventory-update, …;
• distinguish different statuses of data items — Raw, Harbor, Access, ….

Message queues must be durable, for two reasons:

• to allow slow or offline consumers to actually consume the data;
• to guarantee a certain message delivery semantics.

Message delivery semantics

6. The analysis tier: continuous queries

The analysis tier is the heart of the architecture and handles the analysis of the data. Its three key features:

• Adoption of a continuous query model.
• Distributed execution of (possibly idempotent) jobs — idempotent meaning: every time the job receives the same message, it produces the same result; possibly needed when relying on at-least-once semantics.
• Design of algorithms specific to the streaming problem — process items on either a per-item or a per-window basis; rely on approximation for immediate results on complex queries.

The continuous query model

A continuous query is a query that is issued once and then is continuously executed against the data — in contrast, traditional queries are simply executed once when issued. A continuous query may need to maintain a state: an intermediate result that is continuously updated by the query. A query is stateless if each execution is independent from the others.

What makes continuous queries different from traditional ones is two constraints that batch queries never face:

• Memory constraint — for both data processing and state maintenance. Data cannot be processed altogether (one-pass algorithms), and there is not much space to store the state.
• Time constraint — data that cannot be processed in time may have to be dropped (load shedding), and algorithms (e.g., predictive models) may lose efficacy over time (concept drift).

Distributed execution and fault tolerance

The distributed execution of the analysis is an architecture common to different frameworks. Two aspects matter:

• State management: where to store intermediate state. Memory vs disk balances the tradeoff between speed and reliability.
• Fault tolerance: guarantee that data processing doesn’t stop when a worker fails. Two strategies, with opposite resource profiles:
  • Job replication: many nodes carry out the same job. Needed when low latency is a hard requirement (e.g., intrusion detection systems).
  • Rollback recovery: periodically store checkpoints and maintain a log of operations. Lower resource overhead, more expensive recovery. Needed when fault tolerance is important and rare moderate latencies are acceptable.

7. Sliding windows

Windowing techniques allow analyses on a per-window basis instead of a simple per-item basis. In general, windows are defined by two parameters:

• the length, in terms of contained items (sometimes referred to as the eviction policy);
• the period, after which the items in the window are processed (sometimes referred to as the trigger policy).

Sliding windows define both length and period in terms of stream time: items are assigned to windows based on the time at which they are collected. Depending on the comparison of length and period, three specialized versions appear:

8. Data-driven windows and stream time

Data-driven windows define the length in terms of the content that comes with the data: items are assigned to windows based on the values of some fields or attributes in the item itself. In this case, the period determines the update interval rather than the window extent.

The typical use case is sessions:

• A session is a sequence of events terminated by a gap of inactivity greater than some timeout.
• Goal: determine the average amount of traffic generated by sessions.

Sessions bring two problems, and the lecture is careful to attribute each one correctly:

Stream time and event time

Ideally, an event is processed exactly as it happens. In reality, there is a (potentially significant) difference between the time at which an event occurs and the time at which it enters the streaming system:

• Stream time (a.k.a. processing time): defined by the system as the event enters the pipeline.
• Event time: carried by the event itself — the time the event claims to have happened.

The time difference is called skew, and it may be due to hardware issues (network congestions or partitions) or software issues (contention, distributed system logic).

Windowing by stream time

Pros:

• Extremely straightforward implementation.
• Windows are always complete — when the clock says the window is over, it is over.
• It is possible to infer information about the source as it is observed — e.g., tracking the number of requests per second sent to a global-scale web service.

Cons:

• Event time cannot play an important role.
• Even if the data source sends events ordered by event time, it is not guaranteed that they arrive immediately and/or in the same order — you cannot just assume that data arrives in the correct order. The lecture’s examples: GPS data from cars entering a tunnel, traffic from mobile phones going into airplane mode.

9. Windowing by event time: watermarks and late data

Windowing by event time is sometimes referred to as the gold standard of windowing — the reference use case is analyzing user behavior on websites, where the business meaning lives in the event’s own time, not in the server’s. But it is expensive:

• Impossible to precisely know when the window can be closed — there is always the chance that a straggler event for an old window is still in transit.
• Extended window lifetime implies more buffering of data.
• Most processing systems lack native support.

Systems that do support it must therefore define four things: watermarks, triggers, allowed lateness, and an accumulation strategy.

Watermarks

The watermark captures the progress of event-time completeness as processing time progresses. In practice, it is the amount of time that you need (or want) to wait to receive the data produced in a given time. Two kinds:

• A perfect watermark guarantees no late data ever arrives — but it is impractical: it guarantees completeness at the expense of latency.
• A heuristic watermark estimates the progress based on information available about the input stream — faster and more realistic; usually defined as a fixed amount of time; it has to account for possible late data.

Triggers, allowed lateness, accumulation

Triggers declare intervals of intermediate processing of windows. Each partial result determines a pane of the window. Triggers can be based on:

• watermark progress;
• processing time progress;
• element counts;
• record features (e.g., an EOF marker).

Since late data is realistic, a policy for allowed lateness must be defined — e.g., allow data that is 5 minutes late w.r.t. the watermark, and discard everything else. The higher the tolerance, the longer data must be buffered.

The accumulation strategy defines how intermediate results must be aggregated:

10. One-pass algorithms with (ε, δ) guarantees

Consider n, the space of events captured by the data stream. Streaming algorithms have the following requirements:

• One-pass — once examined, items must be discarded;
• Small space for the internal state: O(polylog(n));
• Fast update of the internal state: O(1) to O(polylog(n));
• Fast computation of answers;
• Approximated answers with (ε, δ)-guarantees: let R be the exact result, with ε, δ > 0; with probability at least 1 − δ the algorithm outputs R′ such that (1 − ε)R ≤ R′ ≤ (1 + ε)R.

For instance, take ε = 0.02 and δ = 0.01: then you have a 99% probability that the obtained result R′ equals the real result ± 2%.

What can and cannot be done in one pass? The lecture gives two lists, and the boundary between them is examinable:

Analyzing a data stream is challenging due to space constraints, time constraints, and algorithmic unfeasibility in one pass. One solution is to rely on approximated algorithms, which achieve approximation through two main concepts:

11. Reservoir sampling

Random sampling is a common technique in data streaming. The lecture’s use case is YouTube: 5 million videos are watched every minute (as of 2019), and the goal is a real-time statistical analysis of the watched videos. On the surface that seems pretty easy — but the data moves fast, never stops, and doesn’t fit into memory. A viable solution is to sample the stream as it is flowing. Two questions follow:

• How do we take a random sample from a data set that you can’t hold in memory or on disk?
• How do we know it’s random?

The classical answer is reservoir sampling (J.S. Vitter, 1985, “Random sampling with a reservoir”, ACM Transactions on Mathematical Software 11(1), 37–57), based on the notion of a reservoir: we hold a predetermined number of stream values. When a new value arrives, we probabilistically determine whether to add it to our collection (replacing an occupied slot) or to discard it.

The algorithm

1. The reservoir is always filled with the first r values in the stream, where r is the size of the reservoir.
2. When the i^th data item arrives, it is added to the reservoir with probability P = r / i.
3. If it is added, one of the items already in the reservoir is discarded, each with the same probability (uniform eviction).

In the lecture’s example with r = 15, the 16^th item is added with P = 15/16: a random number α between 0 and 1 is generated; if α > 15/16 the item is discarded, if α ≤ 15/16 it is added and a random resident is evicted. The probability calculus checks out — the lecture works it for the 1^st element across rounds: its probability of being discarded at step 16 is 1/16, and at step 17 it becomes 2/17 = 1/16 + (15/16)·(15/17)·(1/15), where the second term covers “was kept, the 17^th item was chosen for insertion, and the 1^st was chosen for removal”.

The proof, in one line

Consider x_i the i^th element and x_o an older element. For i ≤ r, P(x_i to be put in r) = 1. For i > r:

P(xi to be put in r)    = r/i
P(xo to be already in r) = r/(i-1)
P(xo to be kept in r) =
    P(xo already in) * ( P(xi not put) + P(xi put) * P(someone else evicted) )
  = r/(i-1) * ( (1 - r/i) + (r/i) * ((r-1)/r) )
  = r/(i-1) * (i-1)/i
  = r/i

12. Counting distinct elements: HyperLogLog

The count distinct problem, with the lecture’s use case: Facebook wants to know how many users are connected to my website? from a connections table with (ConnID, Device, UserID) rows.

• Option 1: query the table — select count(distinct UserID) from tab_conn; — O(n) space and O(n) query complexity.
• Option 2: maintain an in-memory hash table of seen UserIDs — O(n) space, but O(1) query complexity.

Both are O(n) in space, and n is billions. The general requirement of both approaches to the count-distinct problem: use a hash function H to translate input items to a uniform distribution in some bounded domain, and study the hash values without storing them.

HyperLogLog (Flajolet, Fusy, Gandouet & Meunier, 2007, “Hyperloglog: the analysis of a near-optimal cardinality estimation algorithm”, Discrete Mathematics and Theoretical Computer Science proceedings) rests on one observation: the cardinality of a set of uniformly distributed random numbers can be estimated by calculating the maximum number of leading (or trailing) zeros in the binary representation of each number in the set. Uniformly distributed random numbers are obtained from the input elements by applying a well-studied hash function.

Why leading zeros work: given a binary string, the probability of observing 1 initial zero is 1/2 (50%), 2 initial zeros is 1/4 (25%), 3 initial zeros is 1/8 (12.5%), and n initial zeros is 1/2ⁿ. The number of attempts required on average to generate a binary string with n trailing zeros is 2ⁿ. So if the maximum number of trailing zeros ever observed is n, it must have taken approximately 2ⁿ attempts.

The space requirement falls out of the arithmetic. With 32-bit hashes: the maximum number of initial zeros is 32 = log₂(2³²), and the number of bits required to store “32” is 5 = log₂(32) = log₂(log₂(2³²)). Hence O(log log n) space.

Considerations

• Very little space. Take N_max = estimated upper bound of the cardinality = 2³² ≈ 4 billion. Storing the raw data would need 4 billion 32-bit hashes ≈ 16 GB; each bucket needs log₂(log₂(N_max)) + O(1) = 5 bits.
• More buckets → more precision. 2048 buckets → 1.5 KB (32-bit hashes, first 11 = log₂(2048) bits identify the bucket), with 98% accuracy.
• The error formula: ε = k · (1.04 / √m), with confidence level 1 − δ = 2Φ(k) − 1. In the lecture’s table: 68.3% confidence → k = 1 → 1.04/√m; 95% → k = 2 → 2.08/√m; 99.7% → k = 3 → 3.12/√m.
• HyperLogLog is distributable: each node keeps its own register, then the registers are merged into one — a property that maps directly onto the cluster architectures of Chapters 5–8.

13. Membership: Bloom filters

The membership problem, with the lecture’s use case: YouTube has 1 billion videos — has a specific video been viewed today?

• Option 1: a hash table of the videos viewed today — 64 bytes per ID × 2³⁰ videos ≈ 69 GB.
• Option 2: an array of n bits with a perfect hash function — 2³⁰ bits ≈ 134 MB, but perfect hash functions are extremely hard to implement.

The old answer is the Bloom filter, conceived by Burton Howard Bloom in 1970 (“Space/time trade-offs in hash coding with allowable errors”, Communications of the ACM 13(7), 422–426). The one property that makes it a streaming tool:

A Bloom filter is an array of m bits, with m > n (n = the number of items to keep track of — the videos). Each bit is a flag: 0 = not seen, 1 = seen. If m = 10n, the space occupation is about 1.3 GB for a billion items — but the deck’s punchline is better stated as: it is 10 bits per element, nothing more, whereas storing the keys would most probably require a lot more space.

Incoming items are processed by a series of hash functions: given item i, the bit h(i) is turned to 1 for every hash function. Ultimately, the answer to “has i been seen?” is given by AND(h₁(i), …, h_k(i)): all k bits must be 1.

Two sizing rules follow from the desired probability of false positives p:

• The optimal number of hash functions k depends on p — for p = 1%, k ≈ 7.
• The optimal number of bits m depends on p and on the expected number of elements n — for p = 1% and n = 10⁹, m = 10¹⁰, i.e. 10 bits per element.

14. Frequency: Count-Min Sketch

The frequency problem, with the lecture’s use case: YouTube has 4 billion views per day — how many times has a specific video been viewed today? Again we want a space-efficient structure, and again the answer relies on hashes and builds on the concept of the Bloom filter. The most common algorithm is the Count-Min Sketch (Cormode & Muthukrishnan, 2005, “An improved data stream summary: the count-min sketch and its applications”, Journal of Algorithms 55.1, 58–75). Its name is its strategy: you first do a series of approximated counts, then you keep the minimum of those.

Setup and query

• Take a set of d hash functions; each function is associated to an array of w counters — a d × w grid.
• Given item i, each hash function increments the counter at position h(i).
• Ultimately, count(i) ≈ min(h₁(i), h₂(i), h₃(i), h₄(i)) — the minimum over the d rows.

Guarantees and sizing

Count-Min Sketch provides (ε, δ)-guarantees. Given the estimated result R′, the true result R and the number of elements n:

• R′ ≥ R (never undercounts), and
• R′ ≤ R + ε·n, with the probability of the previous statement being false equal to δ.

Sizing follows: the number of hash functions d (i.e., rows) depends on δ — for δ = 1%, d = 5. The number of counters w (i.e., columns) depends on ε — for ε = 1%, w = 272. The deck’s worked numbers for n = 1 billion videos: a table for the perfect count would need 64-byte IDs plus 4-byte counters ≈ 73 GB; the sketch with ε·n = 100 (i.e., ε = 9.3×10⁻⁸, w = 29M) is 5 arrays of 29M counters of 4 bytes each → 580 MB.

Usage

The Count-Min Sketch can be used to answer several questions beyond the point query:

• Point queries — how many times has X_i occurred? (With the turnstile model it becomes a median-count.)
• Range queries — how many times have [X₃, …, X₁₀] occurred? (Some extra setup is required.)
• Inner product queries — what is the join size of two streams X and Y? (X and Y collect different data for the same items; both streams must use the same parameters and the same hash functions.)
• Heavy hitters — which items have a multiplicity exceeding a fraction φ of the total cardinality? (A second supporting data structure is needed.)

15. Choosing the streaming tools

The lab deck opens with two practical questions, one per tier of Plate 10.1: which tool for the message queueing tier, and which for the analysis tier. The answers are worth knowing as a landscape, because the rest of the chapter uses two of them.

16. Apache Kafka in practice

Usually you start from a simple interprocess communication channel — e.g., an application that sends some monitoring info to be displayed on some dashboard. But it can get out of hand in no time. The solution is a publish-subscribe messaging system: it reduces the complexity of the architecture, decouples applications, and lets the broker be specialized to handle multiple queues.

Data units: messages and batches

The unit of data within Kafka is called a message — similar to a row or a record. A message is optionally associated to a key (messages with the same key will end up in the same place), and its content can be in any format: JSON, XML, Avro. Messages are transferred in groups, called batches: batches contain messages bound for the same destination, their size is tunable — a direct tradeoff between latency and throughput — and they are typically compressed.

Topics and partitions

Messages are categorized into topics (i.e., the queues) — similar to a database table or a file system folder. Topics are broken down into partitions:

• Partitions provide scalability and redundancy.
• The key of the message determines its partition.
• Ordering of messages is guaranteed only within partitions.

Retention, consumers and offsets

• Message retention. Each topic is committed (i.e., persisted) to disk; the retention policy is tunable (e.g., time-to-live, max capacity). Consumers are able to read only committed messages — if replication is enabled, the message must be committed in every partition. Choosing a serialization format such as Avro enables faster I/O on disk.
• Consumer groups. Consumers are organized into groups; each partition is owned by (i.e., assigned to) only one consumer per group. The group enables scalability and fault tolerance of reads (although ownership rebalancing requires downtime), and more groups can read from the same topic.
• Offsets. Consumers maintain an offset, the last message read; each consumer maintains its own. Offsets can be committed (communicated to the broker, which saves them to disk); offset commits enable failover between consumers in the same group.
• Replication. The replication of partitions enables fault tolerance only — writes and reads are carried out on the leader partition; replicas exist to take over, not to share load.

Delivery semantics in Kafka

The three semantics of section 5 map onto Kafka in a specific way:

• Exactly once — directly supported only when the consumer is a Kafka Streams application; with different consumers it is implementable through a careful management of the offset.
• At least once — supported by default.
• At most once — supported by tuning some configuration parameters: disabling retries on the producer (don’t produce it more than once) and committing offsets in the consumer prior to processing a batch of messages (don’t consume it more than once).

17. Spark Streaming: DStreams and the StreamingContext

Spark Streaming is an extension of the core Spark API. It divides the streaming data into mini-batches: the high-level abstraction is called a discretized stream or DStream, and a DStream is represented as a sequence of RDDs. Everything Chapter 6 taught about RDDs applies to each batch — lineage, lazy evaluation, shuffles — with one second (or whatever interval you choose) of incoming data per RDD.

The StreamingContext object is the main entry point of all Spark Streaming functionality:

import org.apache.spark.streaming._
val ssc = new StreamingContext(sc, Seconds(1))

Note the connection to section 7: mini-batches are fixed windows with length = period = 1 second. Spark Streaming does not start from per-event processing; it starts from the simplest window of all.

After a context is defined, the job is structured as follows:

1. Define the input source(s).
2. Define the streaming computations.
3. Start receiving data and processing it using streamingContext.start().
4. Wait for the processing to be stopped (manually or due to any error) using streamingContext.awaitTermination(); or stop manually using streamingContext.stop().

Input sources and the DStream

Spark Streaming provides two categories of built-in streaming sources:

• Basic sources, directly available in the StreamingContext API — e.g., file systems, socket connections.
• Advanced sources, available through extra utility classes — e.g., Kafka, Flume, Kinesis.

Each RDD in a DStream contains data from a certain fixed interval, and any operation applied on a DStream translates to operations on the underlying RDDs. DStreams support many (though not all) of the transformations available on RDDs … plus some new operations, which are the subject of the next section:

• updateStateByKey — accumulate values across mini-batches;
• transform — apply arbitrary RDD operations;
• window operations;
• join between DStreams.

18. Spark Streaming: state, windows and output

By default, mini-batches are independent of each other: each one is computed as if the previous ones had never happened. To accumulate values across mini-batches, Spark Streaming introduces a concept of state: updateStateByKey associates a state to every key, and for every new mini-batch the state of a key is updated according to a specified function func.

def updateFunction(newValues: Seq[Int], oldValue: Option[Int]): Option[Int] = {
  // returns the new state for the key
}
val cumulativeDs = currentDs.updateStateByKey(updateFunction)

Checkpointing

In order to maintain the state, checkpointing must be set up: a streaming application must operate 24/7, and checkpointing makes it resilient to failures unrelated to the application logic (e.g., system failures, JVM crashes). Two types of data can be checkpointed:

• Metadata (e.g., configuration options, DStream operations, incomplete batches) — required to recover the application from driver failure;
• Data — required if stateful operations are used.

ssc.checkpoint("hdfs:/user/egallinucci/streaming/checkpoint")

Transform

transform is used to apply any RDD operation that is not exposed in the DStream API — e.g., sort, or join with an RDD:

val sortedDs = ds.transform({ rdd => rdd.sortByKey() })

Window operations

Each mini-batch corresponds to a fixed window — the last X seconds of data every X seconds. Other kinds of windowing (sampling and overlapping, from section 7) are obtained by creating a new DStream from the existing one and specifying:

• the length (e.g., 3 seconds);
• the period (e.g., 2 seconds), called slideInterval.

For example: the last 3 seconds of data every 2 seconds — an overlapping window. Both parameters must be multiples of the source DStream’s interval.

Output

Output operations actually allow the transformed data to be consumed by external systems. Similarly to actions for RDDs, they trigger the actual execution of all the DStream transformations. Available operations:

• print() — prints the first ten elements of every batch of data in a DStream on the driver node;
• saveAsTextFiles(prefix, [suffix]), saveAsObjectFiles(prefix, [suffix]), saveAsHadoopFiles(prefix, [suffix]);
• foreachRDD(func) — the most generic output operator: applies a function to each RDD generated from the stream; the function func is executed in the driver process.

19. Structured Streaming: unbounded tables

Spark 2 introduced Structured Streaming to work with DataFrames and Datasets in the streaming scenario — the SQL machinery of Chapter 7 applied to streams. Its conceptual move is the unbounded table: every record from the stream is appended to a table, and the streaming query is a normal relational query over that ever-growing table.

Because the table never finishes, the query must declare how its results leave the system. Structured Streaming defines three output modes:

Test your knowledge