Imagine you are tasked with analyzing a massive CSV file containing millions of records—too large to load into memory all at once. Your goal is to extract only those rows where a specific column's value exceeds a given threshold, saving the filtered results to a new file. This scenario is common in large-scale data analysis, where efficient, memory-friendly processing is essential.


import unittest
import user_code
import ast
import re   
import importlib
import csv
import unittest
import pandas as pd
import os
import csv
import importlib

class TestTask(unittest.TestCase):
    def setUp(self):
        self.input_file = "test_large_input.csv"
        self.output_file = "test_filtered_output.csv"
        self.data = [
            ["id", "value", "name"],
            [1, 50, "Alice"],
            [2, 200, "Bob"],
            [3, 150, "Charlie"],
            [4, 90, "David"],
            [5, 300, "Eve"]
        ]
        with open(self.input_file, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerows(self.data)

    def tearDown(self):
        if os.path.exists(self.input_file):
            os.remove(self.input_file)
        if os.path.exists(self.output_file):
            os.remove(self.output_file)

    def test_rows_filtered_correctly(self):
        import user_code
        importlib.reload(user_code)
        user_code.filter_large_csv(self.input_file, self.output_file, "value", 100, chunk_size=2)
        with open(self.output_file, newline="") as f:
            reader = csv.reader(f)
            rows = list(reader)
        expected = [
            ["id", "value", "name"],
            ["2", "200", "Bob"],
            ["3", "150", "Charlie"],
            ["5", "300", "Eve"]
        ]
        _dynamic_test(
            self,
            rows == expected,
            "Rows with value > 100 are correctly filtered and written to output.",
            f"Expected rows: {expected}, got: {rows}"
        )

    def test_header_only_when_no_match(self):
        import user_code
        importlib.reload(user_code)
        user_code.filter_large_csv(self.input_file, self.output_file, "value", 1000, chunk_size=2)
        with open(self.output_file, newline="") as f:
            reader = csv.reader(f)
            rows = list(reader)
        expected = [["id", "value", "name"]]
        _dynamic_test(
            self,
            rows == expected,
            "Only header is written when no rows match the filter.",
            f"Expected only header row, got: {rows}"
        )

    def test_header_is_written(self):
        import user_code
        importlib.reload(user_code)
        user_code.filter_large_csv(self.input_file, self.output_file, "value", 49, chunk_size=2)
        with open(self.output_file, newline="") as f:
            reader = csv.reader(f)
            rows = list(reader)
        _dynamic_test(
            self,
            len(rows) > 0 and rows[0] == ["id", "value", "name"],
            "Header row is present in the output file.",
            f"Header row missing or incorrect: {rows[0] if rows else 'No rows'}"
        )

    def test_various_chunk_sizes(self):
        import user_code
        importlib.reload(user_code)
        for chunk_size in [1, 2, 5]:
            user_code.filter_large_csv(self.input_file, self.output_file, "value", 100, chunk_size=chunk_size)
            with open(self.output_file, newline="") as f:
                reader = csv.reader(f)
                rows = list(reader)
            expected = [
                ["id", "value", "name"],
                ["2", "200", "Bob"],
                ["3", "150", "Charlie"],
                ["5", "300", "Eve"]
            ]
            _dynamic_test(
                self,
                rows == expected,
                f"Function works with chunk_size={chunk_size}.",
                f"With chunk_size={chunk_size}, expected: {expected}, got: {rows}"
            )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    changed = False
    # Collect all assignment nodes to modify
    assign_nodes = [
        (i, node)
        for i, node in enumerate(tree.body)
        if isinstance(node, ast.Assign)
        and any(isinstance(target, ast.Name) and target.id == var_name for target in node.targets)
    ]

    # If nothing to change, return unmodified code
    if not assign_nodes:
        return code

    # Perform replacements for all matching assignments (from last to first to not break line offsets)
    for i, node in reversed(assign_nodes):
        start_line = node.lineno - 1
        line = lines[start_line]
        indent = ' ' * (len(line) - len(line.lstrip()))
        lines[start_line] = f"{indent}{var_name} = {value}"
        next_line = len(lines)
        for next_node in tree.body[i+1:]:
            if hasattr(next_node, 'lineno'):
                next_line = next_node.lineno - 1
                break
        if next_line > start_line + 1:
            lines[start_line+1:next_line] = []
        changed = True

    return '\\n'.join(lines) if changed else code

if __name__ == "__main__":
    unittest.main()


test_main.py

A practical, hands-on course for aspiring data scientists ready to tackle real-world large data challenges. Learn to efficiently process, sample, and analyze massive datasets using Python and essential libraries. Each section features engaging video explanations and interactive challenges to build your expertise.

Learn foundational strategies for handling datasets too large to fit in memory, including chunking and streaming techniques.

Explore methods to balance and sample large datasets, including oversampling and undersampling.

Learn to use the polars library for fast, memory-efficient data processing.

Challenge: Filtering Large Datasets

Soluzione