Speed Up Your Azure Event Hub Ingestion with Python ThreadPoolExecutor

A practical guide to moving from sequential to parallel processing when sending data to Azure Event Hub

The Problem

Imagine you have a streaming pipeline that processes flight data in real time. For every incoming batch, you need to:

1. Call an external REST API for each flight record.
2. Send the API response to Azure Event Hub.

When you process flights one by one, each call waits for the previous one to finish. The API call alone can take 200 to 500 milliseconds, and the Event Hub send adds another 50 to 100 milliseconds. If you have 200 flights in a batch, that is roughly 60 to 120 seconds of idle waiting, just because everything runs in a straight line.

This is the classic I/O-bound bottleneck. Your CPU is not busy computing anything. It is just waiting for network responses.

The Sequential Approach (Before)

Here is a simplified version of the sequential pattern many teams start with:

import json
import requests
from azure.eventhub import EventHubProducerClient, EventData

def get_api_response(flight_number, departure_date):
    body = [{"flightNumber": flight_number, "departureDate": departure_date}]
    response = requests.post(API_URL, headers=HEADERS, json=body, timeout=30)
    response.raise_for_status()
    return response.json()

def send_to_eventhub(message, conn_str):
    producer = EventHubProducerClient.from_connection_string(conn_str=conn_str)
    batch = producer.create_batch()
    batch.add(EventData(json.dumps(message, default=str)))
    producer.send_batch(batch)
    producer.close()

# Sequential processing
def process_flights_sequential(flight_records):
    for record in flight_records:
        api_data = get_api_response(record["flight_number"], record["departure_date"])
        if isinstance(api_data, list):
            for item in api_data:
                send_to_eventhub(item, EVENTHUB_CONN_STR)
        else:
            send_to_eventhub(api_data, EVENTHUB_CONN_STR)

This works, but it does not scale. Each flight waits in line for the one before it.

The Parallel Approach (After)

Python’s concurrent.futures.ThreadPoolExecutor lets you run multiple I/O operations at the same time using a pool of threads. Since each thread independently handles one API call and one Event Hub send, they all execute concurrently without waiting for each other.

Get Rahul Gosavi’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Here is the improved version:

from concurrent.futures import ThreadPoolExecutor, as_completed

MAX_WORKERS = 50  # Tune this based on your API rate limits and cluster resources

def process_single_flight(flight_number, departure_date):
    """Fetch data from API and send to Event Hub for one flight."""
    try:
        api_data = get_api_response(flight_number, departure_date)
        if isinstance(api_data, dict) and "error" in api_data:
            return (flight_number, departure_date, False)

        if isinstance(api_data, list):
            for item in api_data:
                send_to_eventhub(item, EVENTHUB_CONN_STR)
        else:
            send_to_eventhub(api_data, EVENTHUB_CONN_STR)

        return (flight_number, departure_date, True)
    except Exception as e:
        return (flight_number, departure_date, False)


def process_flights_parallel(flight_records):
    """Process all flights concurrently using a thread pool."""
    success_count = 0
    error_count = 0

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {
            executor.submit(
                process_single_flight,
                rec["flight_number"],
                rec["departure_date"]
            ): rec
            for rec in flight_records
        }

        for future in as_completed(futures):
            flight_number, departure_date, success = future.result()
            if success:
                success_count += 1
            else:
                error_count += 1

    print(f"Done. Total={len(flight_records)}, Success={success_count}, Errors={error_count}")

Sequential vs Parallel — What Changed

Sequential Processing

• Processes one flight at a time, one after another.
• For 200 flights at roughly 300ms per call, the total time is about 60 seconds.
• CPU usage stays low because it is mostly waiting on network I/O.
• Code is simple — just a basic for loop.
• If one call fails, it can break the entire loop unless you add extra error handling.

Parallel Processing (ThreadPoolExecutor)

• Processes up to N flights at the same time using a pool of threads.
• For 200 flights with 50 workers, the total time drops to about 2 to 4 seconds.
• CPU usage is still low, but the wait times now overlap across threads.
• Code is slightly more involved, but still straightforward with Python’s built-in library.
• Each flight runs independently, so one failure does not affect the others.

The key insight is that ThreadPoolExecutor does not make individual calls faster. It simply overlaps the wait times. If each call takes 300ms and you run 50 in parallel, you finish 50 calls in roughly 300ms instead of 15 seconds.

How It Fits in a Spark Streaming Pipeline

In a typical Spark Structured Streaming setup, you receive micro-batches of data using foreachBatch. Inside each batch, you collect the distinct records to the driver node and then fan out the API calls using the thread pool.

def process_streaming_batch(df, epoch_id):
    # Your transformation logic here
    transformed_df = transform(df)

    # Collect distinct flight identifiers to the driver
    flight_records = (
        transformed_df
        .select("flight_number", "departure_date")
        .distinct()
        .collect()
    )

    # Fan out API calls and Event Hub sends in parallel
    process_flights_parallel(flight_records)

# Start the stream
stream = (
    parsed_df.writeStream
    .foreachBatch(process_streaming_batch)
    .outputMode("append")
    .option("checkpointLocation", "/path/to/checkpoint")
    .start()
)

This pattern keeps the heavy data processing in Spark (distributed), while the external API calls happen on the driver using threads (concurrent).

Tips for Tuning

• Start with a conservative worker count. Begin with 10 to 20 workers and increase gradually. Monitor your API provider’s rate limits.
• Set a request timeout. Always use a timeout on your HTTP calls (e.g., timeout=30). A single hanging request should not block the entire pool.
• Track success and failure counts. The as_completed pattern gives you per-record visibility. Log failures so you can retry or alert.
• Reuse the Event Hub producer. In production, consider creating one EventHubProducerClient and sharing it across threads instead of creating a new one per call. This reduces connection overhead.
• Watch your driver memory. Since collect() brings data to the driver, make sure the batch sizes are reasonable.

Conclusion

Switching from a sequential loop to ThreadPoolExecutor for I/O-bound operations like API calls and Event Hub sends is one of the simplest and most impactful optimizations you can make in a data pipeline. The code change is minimal, but the throughput improvement is dramatic.

If your pipeline spends more time waiting on network calls than actually processing data, threading is likely your quickest win.

Thanks for reading. If this helped you speed up your pipeline, feel free to share it with your team.