How to Build Data Processing Functions in Python
Create reusable helper functions to load, filter, transform, and aggregate CSV data in Python.
Python code
45 linesimport csv
from pathlib import Path
def load_data(filepath):
"""Load CSV data into a list of dicts."""
with open(filepath, "r", newline="", encoding="utf-8") as f:
return list(csv.DictReader(f))
def filter_rows(rows, column, value):
"""Keep rows where column equals value."""
return [row for row in rows if row.get(column) == value]
def transform_column(rows, column, func):
"""Apply func to a column and return new rows."""
return [{**row, column: func(row[column])} for row in rows if column in row]
def aggregate(rows, group_key, value_key):
"""Sum value_key grouped by group_key."""
result = {}
for row in rows:
result[row[group_key]] = result.get(row[group_key], 0) + int(row[value_key])
return result
if __name__ == "__main__":
sample = """name,department,salary
Alice,Engineering,100000
Bob,Engineering,90000
Carol,Sales,80000
Dave,Sales,85000
"""
Path("employees.csv").write_text(sample, encoding="utf-8")
data = load_data("employees.csv")
print("Loaded rows:", len(data))
engineers = filter_rows(data, "department", "Engineering")
print("Engineers:", len(engineers))
salaries = aggregate(data, "department", "salary")
print("Salary sums:", salaries)
Output
Loaded rows: 4
Engineers: 2
Salary sums: {'Engineering': 190000, 'Sales': 165000}
How it works
These helper functions form a mini ETL pipeline: load, filter, transform, and aggregate. Using csv.DictReader converts each row into a dictionary—keys are column headers and values are cell contents. List comprehensions with .get() avoid KeyError when columns are missing. The aggregate function uses a simple pattern: result.get(key, 0) + value to accumulate sums. Writing the CSV with Path.write_text makes the example self-contained and reproducible.
Common mistakes
- Forgetting `newline=""` when opening CSV files (causes `\r\n` issues)
- Not handling missing keys with `.get()` in filter and transform
- Assuming empty rows don't exist — filter or validate them early
- Applying transform without checking if column exists
Variations
- Use `csv.DictWriter` to write the transformed rows back to a new file
- Convert to pandas DataFrame and use `groupby().sum()` for larger datasets
Real-world use cases
- ETL jobs that read daily export CSVs, filter for active records, and aggregate totals by region.
- Data migration tasks where you normalize inconsistent column values before loading into a database.
- Reporting scripts that summarize sales, revenue, or usage metrics from flat export files.
Sponsored
More from Data pipelines & processing
- Add a UUID Surrogate Key to Each Row in a CSV with Python easy
- Attach Source File Metadata to Records in Python easy
- Build a Python Utility That Detects Duplicate Records Across Multiple Excel Sheets medium
- Check Null Rate Threshold in PySpark DataFrame medium
- Count Records Processed per Category in Python easy
- Create Data Helper Functions in Python for Beginners easy
Keep learning
Related tutorials and quizzes for this topic.