Data Stream

Processing

Zachary G. Ives
University of Pennsylvania

CIS 5450 – November 20, 2025

© Z. Ives, R. Marcus, J. Gardner, S. Davidson

 Transforming and Responding to
Timeseries: Data Streams
• We’ve discussed learning for time series, but not how to do this in
real-time
• Especially if we want to filter, transform, and integrate data…

• In the past we looked at exploratory queries where we are trying to

discover something new over static data
• But many big data circumstances involve known queries over changing
data – perhaps doing classification or regression (and occasionally
forecasting)
• … and we typically want to do this with low latency
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 2
 Real-Time Stream Processing
new new
data result
• “Real-time” information is arriving across a network
query
• Computation may:
• Involve one object at a time (e.g., predict a Google query needs a visualization)
• Span a group of entries (e.g., count how many tweets seen so far on a topic)
• Involve lookups (e.g., keyword match against known countries)
• Integrate machine learning inference, calls to fetch data from the web, etc.

• In fact, we can capture the computation as a query using ideas from SQL,
Pandas, or Spark (possibly calling user-defined functions, eg for prediction)
• The query gets recomputed each time we receive new data!
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 3
 The Roadmap

• We’ll talk about how we can formalize incremental processing of “data

streams” so we can reuse many of the query ideas we’ve already studied

• We’ll see high-level streaming systems with a focus on Apache Spark

Streaming

• We’ll then look under the covers at lower-level streaming systems, which
offer more responsiveness and power

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 4

 What Is a Data Stream?

• We’ve studied how to process collections of data in tables – relations

or dataframes are multisets or lists of rows

• Select, project, join, group, etc.

• Streams are a different animal – or are they?
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 5
 An Idea: Data Streams –
Prefixes of Infinite DataFrames
TS Value
• If data is coming in real time, the stream is a finite now 13
4
99
5
23
132
92
prefix of an infinite sequence spanning into the future! 1sec ago 92
4
99
5
23
123

• Think of an “infinite dataframe” where each element has 2sec ago 123
4
99
5
23

a timestamp TS and at least one value 3sec ago 23

4
99
5

4sec ago 5
• We will be tracking (modulo latencies) 4
99

stream_df[stream_df[‘TS’] <= [Link]()] 5sec ago 99

4

6sec ago 4

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright}

Computing over Data Streams
8
 Data Streams,
from a Smartphone

As time progresses,
values are received
GPS on multiple
streams.

Compass Each has a

timestamp but
may be processed
time
a bit later

now now now now now

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 9

 Stream Queries
(Continuous Queries) TS Value

now
now 13
92
123
23
Pose a “continuous query” over the table 1sec
1sec ago
ago
92
123
23
5
SELECT COUNT(*)
2sec
2sec ago
ago 123
23
FROM INPUT_STREAM IS 2sec
2sec ago
ago 5
99
WHERE Cond([Link]) 3sec
3sec ago
ago 23
5
3sec ago 99
3sec ago 4
4sec ago 5
4sec ago 99
Query results update each time the input changes!! 4sec ago 4
5sec ago 99
5sec ago 4
Note the output can be considered a stream too! 6sec ago 4
• Consider: stream of data vs stream of updates
• We can name the results of these queries – views !
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 10
 What Is the Output of a Stream Query?
How do we think about the output of a continuous query over a stream?
Some possible ideas

• A row of timestamps + results (e.g., counts)

• The last result (e.g., count) per key

• A sequence of updates (key, old value --> new value)

We consider all cases, but the first version is the most common
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 11
 Stream Processing
within a Bigger Ecosysem

Stream
Queuing Processing
Server
System
“Batch” Database
Streams (DB) Layer

“Lambda architecture”

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 12

Data Stream Processing Systems
 Goals in Processing Real-Time Data
We sometimes want cumulative operations, e.g.,
Overall average across all of history
But mostly: we want to find trends in data, not just aggregate properties from
the start of time
• We want to do this as fast as possible – so don’t store on disk and then process!

Periodic, rolling, and windowed operations

• Average over last 2 minutes, eg vs last 20 minutes
• Do a feature extraction over the trends of the past 5 minutes, run machine learning
prediction

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 14

 Platforms for Stream Queries
• Apache Spark Streaming
• Apache Flink
• Apache Trident
• Amazon Kinesis Analytics SQL-like programming
abstractions

• Apache Samza
• Apache Storm
• Amazon Kinesis Streams
Amazon Lambda Lower level code:
functions to be called
• … on
data arrival
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 15
 What Does a Distributed Stream
Processing System Do?
Typically:
Ensures that messages sent along the streams do get delivered (sometimes
“at least once”)
May ensure that messages are delivered in order (message queues)

• Processes data as it arrives (modulo some latency)

• In some cases, shards the data

• Can recover from crashes, node failures, etc.

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 16
Stream Windows
 Windows over Streams

If we are doing a computation (join, aggregation, prediction, etc.) – what is the

table we are computing over?
• A finite subset of the stream – a window

• As time passes, we append new tuples to table T

• And we “expire” old tuples as they “fall out” of the window

• Akin to “rolling averages” and similar notions

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 18
 Stream Windows
(“tumbling” based on time)

GPS

time

now now now now now

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 19

 Stream Windows
(“sliding”)

GPS

time

now now now now now

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 20

 Stream Windows
(“sliding + partitions”)

Essentially GROUP BY into different windows

GPS 1 1 1 1

2 2 2

time

now now now now now

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 21

 Tumbling vs Sliding vs Partitions
Tumbling windows make sense when we have discrete periods tracked separately
• e.g. different courses on your schedule

Sliding windows make sense when we are trying to make repeated computations
with some memory
• e.g. to spot trends
• We don’t have discrete event boundaries, so we overlap to catch things that span
boundaries

Partitions make sense when our data represents different conceptual streams
• e.g., events from many users, many orders, etc.
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 22
 Time Isn’t the Only Way of Breaking
Streams into Windows

We can partition by number of rows, e.g., LAST 3 ROWS SLIDE BY 2

• (a bit like CNN strides!)
There’s a more general notion called punctuation that determines events for
“opening” a window and later for “closing” it…
• Open a new window when we see a “new patient” tuple
• Close the window for patient X when we see a “patient checked out” tuple

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 23

 Joins across Windows
Recall that join does
1. a Cartesian product of two sets of elements (e.g., from a
DataFrame, now from a window)
2. a filter based on a predicate (e.g. equality of attributes),
possibly including times As before:
As time passes, we append new
tuples to R and S
Output:
r1s1 if r1[X] = s1[Y]
r1 And we “expire” old tupless1as they
R “fall out” of the
S window
Let timestamp(r1s1)
join = max(timestamp(r1),
timestamp(s1))
R S

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 24

 Stream Window Joins
1 1 1 1 1
Output A B C D E
2 2 2
A C F

1 1 1 1

GPS
2 2 2

Network A B C D E F
Info
time
now now now now now

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 25

 Example of a Stream Join:
Network Activity per User by Day
Join user ID x process ID x process network utilization

[Link]

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 26

 Storage
• In the early days of data stream research: there was an assumption streams are
too fast-moving to store! e.g., like the Large Hadron Collider
• But today: we have found most applications aren’t the LHC
• There is value in storing the results – for auditing and compliance (!), and for
retrospective analysis, forecasting, and training of new models!

• Lots of different approaches to storing time-varying data streams

• Amazon Kinesis Firehose does storage to Amazon’s cloud services
• Apache Flume does logging to HDFS
• Can build adapter from stream engine to storage systems
• Time series databases: Timescale, InfluxDB, …
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 27
 Recap: Stream Windows

• Streams are conceptually infinite, but we typically look “locally” based on

current time
• Stream processing typically goes directly over the data – no storage
• LAST n ELEMENTS, LAST n SECONDS
• Tumbling vs sliding windows
• PARTITIONing windows

• Aggregates are across elements in a (partitioned) window

• Joins are across pairs of windows, rather than pairs of tables
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 31
High Level Stream Processing
 Recall Apache Spark

• Sharded processing of Spark DataFrames (relations)

• What if I have streaming data? Discretized streams, Dstreams
• Today in Spark this is called Structured Streams

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 33

 Streaming in Spark

• Spark “structured streams”: streaming dataframes

• At any point in time, you can do regular SQL / Spark Dataframes operations over
this (also combining with regular dataframes)

• Periodically, Spark can recompute the dataframe based on new data

• Then we need to update the output…

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 34

 Processing Data Streams

Spark query reads from a distributed streaming service

• As data comes in, it gets queued up

Spark Streaming periodically executes…

• Recomputes query output given existing + new data!

Let’s watch a directory for new files and process them as new
data…

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 35

 An Example
Can we build a model to predict airplane flight delays?
• Kaggle competition: [Link]

[Link] [Link]/2018/05/airport-flight-board. html

We’ll start with stream queries that are periodically recomputed

• e.g., we are continuously receiving flight info as status is updated
© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 36
 Streaming Flight Info

flightsStreamDF = [Link]("sep", ",").

option("header", "true").\
option("maxFilesPerTrigger", 1).\
schema(flightSchema).csv("/in/")
[Link]("flights")
Flights(YEAR, MONTH, DAY_OF_MONTH, AIRLINE_ID,
CARRIER, FL_NUM, ORIGIN, DEST, ARR_DELAY, CANCELLED)
Also want Airports(IATA_CODE,AIRLINE,AIRPORT,CITY,STATE,
COUNTRY,LATITUDE,LONGITUDE)
Airlines(IATA_CODE,AIRLINE)

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 37

 Regular Dataframes
for Static Data…
# The airlines
airlinesDF = [Link]("sep", ",").option("header",
"true").\
schema(airlineSchema).csv('[Link]')
[Link]("airlines")

# The airports
airportsDF = [Link]("sep", ",").option("header",
"true").\
schema(airportSchema).csv('[Link]')
[Link]("airports")

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 38

 A Streaming Spark Query
avg_delay = [Link]("""select CARRIER,FL_NUM, ORIGIN,
DEST, [Link] AS from_lat, [Link] AS from_long,
[Link] AS to_lat, [Link] AS to_long,
count(*) as NbrFlights,
avg(ARR_DELAY) as avg_delay
from flights f join airports org
on [Link]=org.IATA_CODE
join airports dst on [Link]=dst.IATA_CODE
GROUP BY CARRIER, FL_NUM, ORIGIN, DEST,
[Link], [Link], [Link], [Link]
ORDER BY CARRIER, FL_NUM, ORIGIN, DEST""")

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 39

 Making It Run
Every 1 second, recomputes an in-memory table called flight_info with
complete results
query = avg_delay.[Link]("complete")
.queryName("flight_info").format("memory").\
trigger(processingTime='1 seconds').start()

for filename in [Link]('/content'):

if filename[0] == 'x':
[Link](Path('/content/' + filename),Path('/in’))
sleep(3)

copies files into /in every ~3 seconds

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 40

What It Looks Like

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 41

 Summary So Far

• Streaming dataframes are updated by Spark periodically

• Queries can be computed over streaming and traditional dataframes
• We can start a query and have it periodically re-execute
… Now let’s look at doing analysis across time!

© Z. Ives, R. Marcus, J. Gardner, S. Davidsonright} 42