Real-Time Data Streaming with Python – Kafka & WebSockets Guide

Most data science courses teach batch processing — run a job overnight, get results in the morning. But many critical applications — fraud detection, real-time recommendations, IoT monitoring, trading systems — require decisions in milliseconds, not hours. This guide covers real-time data streaming with Python, focusing on Kafka and WebSockets.

Batch vs Stream Processing

Batch processing collects data, processes it all at once, and produces output. It’s simpler and efficient for historical analysis. Stream processing processes data continuously as it arrives, producing output with low latency. Use streaming when: you need to detect events as they happen (fraud, anomalies), you need to update dashboards in real time, or you’re building event-driven microservices. The tools are more complex, but the value of low latency is often worth it.

WebSockets – Simple Real-Time Data

pip install websockets asyncio

import asyncio
import websockets
import json

# Client: receive real-time crypto prices
async def stream_prices():
    uri = "wss://stream.binance.com:9443/ws/btcusdt@trade"
    async with websockets.connect(uri) as ws:
        while True:
            msg  = await ws.recv()
            data = json.loads(msg)
            price = float(data['p'])
            print(f"BTC/USDT: ${price:,.2f}")

            # Real-time scoring
            if price < ALERT_THRESHOLD:
                send_alert(f"Price drop: ${price:,.2f}")

asyncio.run(stream_prices())

WebSocket Server

import asyncio
import websockets
import json

connected_clients = set()

async def broadcast(message):
    if connected_clients:
        await asyncio.gather(*[client.send(message) for client in connected_clients])

async def handler(websocket, path):
    connected_clients.add(websocket)
    try:
        async for message in websocket:
            data = json.loads(message)
            await broadcast(json.dumps({"echo": data, "clients": len(connected_clients)}))
    finally:
        connected_clients.remove(websocket)

async def main():
    async with websockets.serve(handler, "localhost", 8765):
        await asyncio.Future()  # run forever

asyncio.run(main())

Apache Kafka – Production Streaming

Kafka is the standard for high-throughput, fault-tolerant event streaming. It's used by Netflix, Uber, LinkedIn, and thousands of other companies to move billions of events per day.

pip install confluent-kafka

Kafka Producer

from confluent_kafka import Producer
import json, time

producer = Producer({'bootstrap.servers': 'localhost:9092'})

def delivery_report(err, msg):
    if err:
        print(f'Message delivery failed: {err}')

# Produce events
for transaction in transactions:
    producer.produce(
        topic='transactions',
        key=str(transaction['user_id']),
        value=json.dumps(transaction).encode('utf-8'),
        callback=delivery_report)
    producer.poll(0)  # trigger delivery callbacks

producer.flush()  # wait for all messages to be delivered

Kafka Consumer

from confluent_kafka import Consumer

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id':          'fraud-detector',
    'auto.offset.reset': 'earliest',
})
consumer.subscribe(['transactions'])

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None: continue
    if msg.error():
        print(f"Error: {msg.error()}")
        continue

    transaction = json.loads(msg.value().decode('utf-8'))

    # Real-time fraud scoring
    features = extract_features(transaction)
    score    = fraud_model.predict_proba([features])[0][1]

    if score > 0.85:
        flag_for_review(transaction, score)

consumer.close()

Stream Processing with Faust

pip install faust-streaming

import faust

app = faust.App('fraud-detection', broker='kafka://localhost:9092')

transactions_topic = app.topic('transactions', value_type=bytes)
alerts_topic       = app.topic('fraud-alerts', value_type=bytes)

@app.agent(transactions_topic)
async def process_transactions(transactions):
    async for txn_bytes in transactions:
        txn      = json.loads(txn_bytes)
        features = extract_features(txn)
        score    = fraud_model.predict_proba([features])[0][1]

        if score > 0.85:
            alert = {'txn_id': txn['id'], 'score': score, 'user': txn['user_id']}
            await alerts_topic.send(value=json.dumps(alert).encode())

if __name__ == '__main__':
    app.main()

Windowed Aggregations

from collections import deque
from datetime import datetime, timedelta

class SlidingWindowAggregator:
    def __init__(self, window_seconds=300):  # 5-minute window
        self.window  = timedelta(seconds=window_seconds)
        self.events  = deque()

    def add(self, value, timestamp=None):
        ts = timestamp or datetime.now()
        self.events.append((ts, value))
        self._trim(ts)
        return self.stats()

    def _trim(self, now):
        while self.events and now - self.events[0][0] > self.window:
            self.events.popleft()

    def stats(self):
        vals = [v for _, v in self.events]
        return {'count': len(vals), 'mean': sum(vals)/len(vals) if vals else 0,
                'max': max(vals) if vals else 0}

# Track rolling 5-minute transaction velocity per user
agg = SlidingWindowAggregator(window_seconds=300)
for event in event_stream:
    stats = agg.add(event['amount'])
    if stats['count'] > 20:  # more than 20 txns in 5 minutes
        flag_velocity_fraud(event['user_id'])

Conclusion

Real-time streaming is a significant step up in complexity from batch processing, but it unlocks use cases that batch simply can't address. Start with WebSockets for simple streaming needs (live dashboards, price feeds). Move to Kafka when you need durability, replay, multiple consumers, or high throughput. Faust and similar stream processing frameworks add the aggregation and windowing logic that transforms raw event streams into actionable signals. The investment is worth it for any application where seconds of latency have business impact.