Blog

Let’s learn how you can work with data that’s arriving as a stream using Pandas. It’s important to understand upfront that Pandas DataFrames are primarily designed for static datasets that fit into memory. Pandas itself doesn’t have a built-in “streaming” mode like dedicated stream processing frameworks.

However, you can absolutely use Pandas to process data from a stream in chunks or batches. This is the standard way to handle streaming data when you want to leverage Pandas’ powerful data manipulation capabilities.

Here’s the core idea and common approaches:

1. Read Data in Chunks: Instead of trying to load the entire, potentially endless stream into a single DataFrame at once, you read and process the data in smaller, manageable batches.
2. Process Each Chunk: Apply your Pandas operations (cleaning, transformation, aggregation, etc.) to each individual batch (which will be a temporary DataFrame or Series).
3. Manage State (if needed): If you need to maintain aggregations or state across batches (e.g., a running total, min/max values over the entire stream), you’ll need to store and update this state outside of the individual chunk processing.
4. Output/Store Results: Decide what to do with the processed chunks or the aggregated state – save them to a new file, send them to a database, display them, etc.

Common Scenarios and How to Handle Them with Pandas:

Scenario 1: Reading a Large File Incrementally (Simulating a Stream from Disk)

If your “stream” is actually a very large file that won’t fit into memory, you can use Pandas’ built-in chunksize parameter when reading.

import pandas as pd

# Assume 'large_dataset.csv' is a file too big for memory

# Create a dummy large file for demonstration
# (In a real scenario, this file would already exist)
data_lines = ["col1,col2,col3"] + [f"{i},{i*2},{i*3}" for i in range(10000)]
with open('large_dataset.csv', 'w') as f:
f.write('\n'.join(data_lines))

print("Processing data in chunks:")

# Use iterator=True or chunksize to get a TextFileReader object
# This object is an iterator that yields DataFrames (chunks)
chunk_iterator = pd.read_csv('large_dataset.csv', chunksize=1000) # Read 1000 rows at a time

total_rows_processed = 0
# Example: Keep a running sum of col2 across all chunks
running_sum_col2 = 0

# Iterate over the chunks
for i, chunk_df in enumerate(chunk_iterator):
total_rows_processed += len(chunk_df)
print(f"Processing chunk {i+1}, shape: {chunk_df.shape}")

# --- Apply your Pandas operations to the chunk_df ---
# Example: Filter rows where col1 > 500
filtered_chunk = chunk_df[chunk_df['col1'] > 500]
# Example: Update running sum
running_sum_col2 += filtered_chunk['col2'].sum()
# --- End of Pandas operations on chunk ---

# After processing all chunks
print("\nFinished processing all chunks.")
print(f"Total rows processed: {total_rows_processed}")
print(f"Running sum of col2 (where col1 > 500): {running_sum_col2}")

# Clean up the dummy file
import os
os.remove('large_dataset.csv')

In this example:

• pd.read_csv(‘large_dataset.csv’, chunksize=1000) doesn’t load the whole file. It returns an iterator.
• The for loop gets one chunk_df (a standard Pandas DataFrame containing 1000 rows) at a time.
• You perform your analysis/transformations on this small chunk_df.
• We maintain running_sum_col2 outside the loop to aggregate results across chunks.

Scenario 2: Receiving Data from a Network Socket, API, or Message Queue

If your data arrives over a network or from a stream source like Kafka or RabbitMQ, your application will typically receive data piece by piece. You’ll need to:

• Implement the logic to connect to the source and receive data.
• Buffer the incoming data until you have a sufficient batch size.
• Once a batch is ready, load it into a Pandas DataFrame.
• Process the DataFrame batch.
• Repeat.

This requires more custom code to manage the data reception and buffering.

# This is a conceptual example, as setting up a real stream source is complex.
# Imagine 'receive_data_batch()' is a function that blocks until
# it has accumulated 'batch_size' records from your stream source
# and returns them as a list of dictionaries or a similar format.

# from your_stream_library import receive_data_batch # Conceptual import

batch_size = 500
total_records_processed = 0
# Example: Store processed results from each batch
all_processed_data = []

print("Listening for data stream batches...")

try:
while True: # Keep listening
# Get a batch of raw data from the stream source
# This function would handle connection, receiving, and buffering
# raw_batch = receive_data_batch(size=batch_size) # Conceptual call

# --- Simulate receiving a batch ---
# In a real app, raw_batch comes from the stream
import time
time.sleep(1) # Simulate waiting for data
if total_records_processed >= 2000: # Simulate stream ending after 4 batches
print("Simulating end of stream.")
break
raw_batch = [{'id': i, 'value': i*10} for i in range(total_records_processed, total_records_processed + batch_size)]
if not raw_batch: # Stream ended or no data received
# Handle graceful shutdown or retry logic
print("No data received, potentially stream end.")
break
# --- End Simulation ---

# Load the raw batch into a Pandas DataFrame
batch_df = pd.DataFrame(raw_batch)
total_records_processed += len(batch_df)

print(f"Received and processing batch of {len(batch_df)} records. Total processed: {total_records_processed}")

# --- Apply your Pandas operations to the batch_df ---
batch_df['value_squared'] = batch_df['value'] ** 2
# --- End of Pandas operations ---

# Store or output the processed batch results
all_processed_data.append(batch_df)
# Or save to database, send to another service, etc.

except Exception as e:
print(f"An error occurred: {e}")
# Handle errors, logging, potentially reconnecting

# Optionally, combine all processed batches at the end if needed and memory allows
# final_df = pd.concat(all_processed_data, ignore_index=True)
# print("\nCombined processed data (if memory allows):")
# print(final_df.head())

Key Considerations When Using Pandas with Streams:

• Memory: The size of your chunks/batches is critical. It must be small enough to fit comfortably in memory along with your processed data and any state you’re maintaining.
• Processing Speed: Ensure your Pandas operations on each batch are faster than or equal to the rate at which data is arriving. If processing lags behind, your buffer will grow, potentially leading to memory issues or data loss.