Sequential data pipelines leave most of your hardware idle. Learn how to use threads and processes to split workloads across workers, cut pipeline runtimes by 10× or more, and handle failures gracefully — with production-ready Python code you can adapt immediately.
Imagine you're running a data pipeline that pulls sales records from 50 regional databases, transforms each batch, and loads the results into a central data warehouse. You run it on a Monday morning, and four hours later your stakeholders are still waiting. The pipeline finished each database sequentially — one at a time, in a single thread — even though your machine had 16 cores sitting idle at 5% utilization. You just left 95% of your processing power on the table.
This is one of the most common and costly inefficiencies in data engineering. Processing workloads sequentially is the default behavior for most Python scripts, and it feels natural because that's how we read and write code — top to bottom, one step at a time. But data pipelines don't have to work that way. With concurrency and parallelism, you can split a workload across multiple workers and complete in 20 minutes what used to take 4 hours.
By the end of this lesson, you'll understand the conceptual difference between concurrency and parallelism, know when to use each approach, and be able to implement both using Python's standard library. You'll also understand the failure modes that trip up beginners and how to avoid them.
What you'll learn:
ThreadPoolExecutorProcessPoolExecutorYou should be comfortable reading Python code and understand what a data pipeline does at a high level. If you're new to pipelines entirely, start with What is a Data Pipeline? Architecture and Core Concepts for Data Engineers before continuing. You should also have a basic sense of how batch pipelines are structured — take a look at Building Your First Data Pipeline with Python if you haven't already.
Let's make the problem concrete. You have a pipeline that processes daily transaction files for 20 retail clients. Each file takes about 3 seconds to download, 1 second to transform, and 2 seconds to upload to the destination. Total: 6 seconds per client.
Run them sequentially:
clients = ["client_a", "client_b", ..., "client_t"] # 20 clients
for client in clients:
data = download_file(client) # 3 seconds (waits for network)
transformed = transform(data) # 1 second (CPU work)
upload_result(transformed) # 2 seconds (waits for network)
Total runtime: 20 × 6 = 120 seconds.
Notice something: two of those three steps — downloading and uploading — are just waiting. Your CPU is idle while the network does its thing. That wasted waiting time is what concurrency and parallelism are designed to reclaim.
These two terms get used interchangeably, but they mean different things and lead to different solutions.
Concurrency is about dealing with many things at once. Think of a chef who puts a pot of water on to boil, then chops vegetables while the water heats, then checks the pot. The chef is only ever doing one thing at a time, but they're managing multiple tasks by interleaving work with waiting. In Python, this is achieved with threads — they share the same process and take turns executing, switching during I/O waits.
Parallelism is about doing many things at once. Think of a kitchen with four chefs, each cooking a different dish simultaneously. Each chef uses their own hands, their own cutting board. In Python, this requires multiple processes, each with its own Python interpreter and memory space.
Key insight: Python has a notorious limitation called the Global Interpreter Lock (GIL), which prevents multiple threads from executing Python bytecode simultaneously. This means threads in Python are excellent for I/O-bound work (where you're mostly waiting) but do NOT provide true parallelism for CPU-bound computation. For CPU-heavy work, you need multiple processes.
The practical rule is simple:
In most real pipelines, the bottleneck is I/O — waiting for APIs to respond, databases to return rows, cloud storage to accept uploads. This makes thread-based concurrency the workhorse for the majority of pipeline scenarios you'll encounter.
Before you can run work in parallel, you need to split it into independent chunks. This is called workload partitioning. The key word is independent — each chunk must be able to run without needing results from another chunk mid-flight.
Common partitioning strategies for pipelines:
| Partition By | Example |
|---|---|
| Entity (client, tenant, region) | Process each of 50 clients separately |
| Time window | Process each of 30 days separately |
| File or object | Process each file in an S3 folder |
| ID range | Process rows 1–10,000, then 10,001–20,000, etc. |
For our 20-client example, partitioning by client is natural — each client's data is independent.
Warning: Not all work can be partitioned. If step B requires the output of step A, those steps must remain sequential. Parallelism works on the width of your workload, not the depth. A pipeline with five sequential transformation steps doesn't become five times faster with parallelism — but a pipeline that runs the same five steps for 100 separate files absolutely can.
Python's concurrent.futures module provides a clean, high-level interface for both threading and multiprocessing. Let's build the 20-client pipeline with threads.
First, here's the sequential version:
import time
def process_client(client_id: str) -> dict:
"""Simulate downloading, transforming, and uploading one client's data."""
print(f" Starting {client_id}")
time.sleep(3) # Simulates network download
records = [{"client": client_id, "value": i} for i in range(100)]
time.sleep(1) # Simulates CPU transformation
time.sleep(2) # Simulates upload
print(f" Finished {client_id}")
return {"client": client_id, "records_processed": len(records)}
clients = [f"client_{chr(65 + i)}" for i in range(20)] # client_A through client_T
start = time.time()
results = [process_client(c) for c in clients]
elapsed = time.time() - start
print(f"Sequential: {elapsed:.1f}s, processed {len(results)} clients")
Running this takes approximately 120 seconds. Now the concurrent version:
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
def process_client(client_id: str) -> dict:
"""Same function — no changes needed."""
print(f" Starting {client_id}")
time.sleep(3) # Simulates network download
records = [{"client": client_id, "value": i} for i in range(100)]
time.sleep(1) # Simulates CPU transformation
time.sleep(2) # Simulates upload
print(f" Finished {client_id}")
return {"client": client_id, "records_processed": len(records)}
clients = [f"client_{chr(65 + i)}" for i in range(20)]
start = time.time()
with ThreadPoolExecutor(max_workers=10) as executor:
futures = {executor.submit(process_client, client): client for client in clients}
results = []
for future in as_completed(futures):
client = futures[future]
try:
result = future.result()
results.append(result)
except Exception as e:
print(f"ERROR processing {client}: {e}")
elapsed = time.time() - start
print(f"Concurrent: {elapsed:.1f}s, processed {len(results)} clients")
With 10 workers handling 20 clients, you now run two batches of 10 in parallel: approximately 12 seconds instead of 120. That's a 10× speedup.
A few things to notice about this code:
ThreadPoolExecutor(max_workers=10) creates a pool of 10 threads. Work is submitted to the pool as futures — objects that represent a computation that hasn't finished yet.
as_completed(futures) yields each future as it finishes, regardless of submission order. This is important: you don't want to block waiting for client_A to finish before you can collect client_B's result if client_B finished first.
The try/except block per future is critical. If one client fails, you collect the error and continue processing the others. Without this pattern, an unhandled exception from a single future can silently swallow your results or crash the entire collection loop.
Tip: How many workers should you use? For I/O-bound pipelines, a good starting point is 5–20 workers per CPU core, since threads spend most of their time waiting rather than computing. For CPU-bound work with processes, match the number of workers to your CPU core count — going beyond that adds overhead without benefit. Always profile before tuning.
Now let's say your transformation step is genuinely CPU-intensive — you're parsing and validating millions of rows of JSON, running regex-heavy cleaning, or computing cryptographic hashes. Threads won't help here because of the GIL. You need processes.
import time
import json
import hashlib
from concurrent.futures import ProcessPoolExecutor, as_completed
def process_partition(partition_id: int, row_count: int) -> dict:
"""
Simulate CPU-intensive work: hashing a large number of records.
Each partition processes 'row_count' rows.
"""
results = []
for i in range(row_count):
# CPU-bound: computing SHA256 for each record
record = json.dumps({"partition": partition_id, "row": i, "value": i * 3.14})
hashed = hashlib.sha256(record.encode()).hexdigest()
results.append(hashed)
return {"partition": partition_id, "processed": len(results)}
# Imagine 8 partitions of 100,000 rows each
partitions = [(pid, 100_000) for pid in range(8)]
start = time.time()
with ProcessPoolExecutor(max_workers=4) as executor:
futures = {
executor.submit(process_partition, pid, count): pid
for pid, count in partitions
}
results = []
for future in as_completed(futures):
pid = futures[future]
try:
result = future.result()
results.append(result)
print(f"Partition {result['partition']} done: {result['processed']} rows")
except Exception as e:
print(f"ERROR in partition {pid}: {e}")
elapsed = time.time() - start
print(f"Parallel: {elapsed:.2f}s, {len(results)} partitions")
Warning: There are important constraints when using
ProcessPoolExecutor. The function you submit and all of its arguments must be picklable (serializable) because they need to be sent to a separate process via inter-process communication. Lambda functions, database connections, and certain class instances cannot be pickled. If you get aPicklingError, move your function to the module level and pass simple data types as arguments.
Another difference: each worker process has its own memory space. Changes to a shared list inside a worker don't affect the parent process. That sounds like a limitation, but it's actually a feature — it eliminates a whole category of race condition bugs that plague multi-threaded code.
The most common mistake newcomers make with concurrent pipelines is assuming they can safely share mutable state — a list, a dictionary, a counter — across workers.
Here's a broken example:
from concurrent.futures import ThreadPoolExecutor
results = [] # Shared mutable state — DANGER
def process_and_append(client_id: str):
data = {"client": client_id, "records": 42}
results.append(data) # Multiple threads writing simultaneously
with ThreadPoolExecutor(max_workers=10) as executor:
executor.map(process_and_append, [f"client_{i}" for i in range(100)])
print(len(results)) # Might not be 100 — or might crash
Python's list.append() happens to be thread-safe in CPython due to the GIL, but you shouldn't rely on this. Other operations — like results[i] += 1 or writing to a dict — are not atomic and can produce corrupted data under concurrent access.
The clean solution: don't share state at all. Have each worker return its result, and collect them in the main thread:
from concurrent.futures import ThreadPoolExecutor, as_completed
def process_client(client_id: str) -> dict:
return {"client": client_id, "records": 42}
with ThreadPoolExecutor(max_workers=10) as executor:
futures = [executor.submit(process_client, f"client_{i}") for i in range(100)]
results = [f.result() for f in as_completed(futures)]
print(len(results)) # Always 100
This pattern — workers compute and return, the main thread collects — scales cleanly and avoids nearly every shared-state bug.
Note: If you genuinely need a shared counter (say, to track total rows processed across all workers), use
threading.Lock()for threads ormultiprocessing.Valuefor processes. But start by asking whether you can restructure the problem so workers don't need to share state. You usually can.
What you've been building is a specific pipeline design pattern called fan-out/fan-in. The pipeline fans out from a single orchestrating process to many workers, then fans back in to collect and consolidate results. This pattern is so common in data engineering that it has a dedicated name and set of best practices.
When pipelines get more complex — say, you need parallel extraction followed by centralized validation — you'll start seeing these patterns composed together. Data Pipeline Design Patterns: Fan-Out, Fan-In, and Branching Workflows Explained covers these architectures in depth and is worth reading once you're comfortable with the concurrency primitives here.
For larger pipelines that need persistent execution history and retry logic across parallel tasks, orchestrators like Airflow let you declare task dependencies explicitly — check out Scheduling and Orchestrating Pipelines with Airflow for how that works in practice.
Adding more workers doesn't always equal more speed. Consider what happens downstream: if your pipeline fans out to 100 threads all hitting the same PostgreSQL database simultaneously, you might overwhelm the connection pool and cause more failures than you solve.
This concept — where a fast producer overwhelms a slow consumer — is called backpressure. Backpressure, Throughput Tuning, and Bottleneck Diagnosis in High-Volume Data Pipelines goes deep on how to detect and resolve this in production. For now, keep these sizing heuristics in mind:
The best approach is to start conservative (4-8 workers), measure your runtime and error rate, then increase incrementally until you hit diminishing returns or downstream errors.
Tip: When your parallel pipeline interacts with external APIs, build in retry logic per worker so transient failures don't drop records. Pipeline Retry Logic and Exponential Backoff: Handling Transient Failures in Data Pipelines walks through the exact pattern — it combines very naturally with
ThreadPoolExecutor.
In a sequential pipeline, one bad record can crash the whole run. In a concurrent pipeline with proper error handling, a bad partition fails in isolation while all the others succeed. This is one of the underappreciated benefits of the fan-out pattern.
Here's a production-grade collection pattern that separates successes from failures:
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import List, Any
@dataclass
class PipelineResult:
successes: List[dict] = field(default_factory=list)
failures: List[dict] = field(default_factory=list)
def run_concurrent_pipeline(work_items: list, worker_fn, max_workers: int = 8) -> PipelineResult:
result = PipelineResult()
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(worker_fn, item): item for item in work_items}
for future in as_completed(futures):
item = futures[future]
try:
output = future.result()
result.successes.append(output)
except Exception as e:
result.failures.append({
"item": item,
"error": str(e),
"error_type": type(e).__name__
})
return result
# Usage
def fetch_client_report(client_id: str) -> dict:
if client_id == "client_F":
raise ConnectionError("Simulated network failure")
return {"client": client_id, "rows": 1500}
clients = [f"client_{chr(65 + i)}" for i in range(10)]
pipeline_result = run_concurrent_pipeline(clients, fetch_client_report, max_workers=5)
print(f"Succeeded: {len(pipeline_result.successes)}")
print(f"Failed: {len(pipeline_result.failures)}")
for failure in pipeline_result.failures:
print(f" {failure['item']} -> {failure['error_type']}: {failure['error']}")
This gives you a complete picture of what worked and what didn't. From here, you can route failed items to a dead letter queue for later inspection — a pattern covered in Implementing Dead Letter Queues and Poison Message Handling in Data Pipelines.
Here's a realistic scenario to cement your understanding. You have a list of 15 public REST API endpoints (you can simulate them), and your job is to build a concurrent pipeline that:
Starter code:
import time
import random
from concurrent.futures import ThreadPoolExecutor, as_completed
# Simulated API endpoints
ENDPOINTS = [f"https://api.example.com/region/{region}/sales" for region in [
"northeast", "southeast", "midwest", "southwest", "northwest",
"new_england", "mid_atlantic", "great_lakes", "plains", "mountain",
"pacific", "south_central", "north_central", "gulf_coast", "appalachia"
]]
def fetch_and_transform(endpoint: str) -> dict:
"""
TODO: Implement this function.
1. Simulate a network call (sleep 1-3 seconds randomly)
2. Randomly fail ~20% of the time with a ConnectionError
3. Return a dict with: endpoint, record_count (random 100-1000), fetch_time
"""
pass # Your code here
def run_pipeline(endpoints: list, max_workers: int = 5):
"""
TODO: Implement this function.
Use ThreadPoolExecutor to run fetch_and_transform concurrently.
Collect successes and failures separately.
Print a summary at the end.
"""
pass # Your code here
run_pipeline(ENDPOINTS)
Expected output should look something like:
Starting pipeline for 15 endpoints with 5 workers...
[0.8s] Fetched northeast: 742 records
[1.1s] FAILED gulf_coast: ConnectionError
...
Summary: 12 succeeded, 3 failed | Total time: 9.2s | Records: 8,431
Try varying max_workers between 1, 5, and 15 and observe how total runtime changes.
Mistake 1: Using threads for CPU-bound work and wondering why it's not faster.
Threads won't speed up pure Python computation. Profile first — if your CPU is pegged at 100% on one core while the others are idle, switch to ProcessPoolExecutor.
Mistake 2: Creating a new executor inside the worker function. Executors manage a pool of workers. Creating one inside a function that itself runs inside an executor causes explosive resource consumption. Always create your executor in the outermost scope and pass work items in.
Mistake 3: Not handling exceptions per future.
If you use executor.map() instead of as_completed(), exceptions are re-raised when you iterate the results — one failure stops the iteration. Use as_completed() with per-future try/except for production code.
Mistake 4: Assuming order is preserved.
as_completed() yields results in completion order, not submission order. If order matters (e.g., you need to write results to a file in client ID order), collect all results first, then sort.
Mistake 5: Ignoring downstream capacity. Launching 50 threads to write to a single SQLite database file will produce errors and corruption. Match your concurrency to what the destination system can handle.
Concurrency and parallelism transform what a data pipeline can do at scale. The key decisions are:
The patterns you've learned here — thread pools for I/O, process pools for computation, fan-out/fan-in collection, and failure isolation — are the foundation for any high-throughput pipeline you'll build.
From here, you'll want to dig into Data Pipeline Error Handling and Recovery Strategies to build resilience into your parallel workers, and Logging, Alerting, and Observability for Data Pipelines to gain visibility into how your workers are actually performing in production. When your concurrent pipelines grow large enough to need state management across restarts, Checkpointing and State Management in Long-Running Data Pipelines will become essential reading.