Stellen Sie sich vor, Sie sollen eine riesige CSV-Datei mit Millionen von Datensätzen analysieren – zu groß, um sie vollständig in den Arbeitsspeicher zu laden. Ziel ist es, nur die Zeilen zu extrahieren, in denen der Wert einer bestimmten Spalte einen vorgegebenen Schwellenwert überschreitet, und die gefilterten Ergebnisse in einer neuen Datei zu speichern. Dieses Szenario ist typisch für die Analyse großer Datenmengen, bei der effiziente und speicherschonende Verarbeitung entscheidend ist.


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

Ein praxisorientierter Kurs für angehende Data Scientists, die bereit sind, reale Herausforderungen mit großen Datenmengen zu meistern. Lernen Sie, massive Datensätze effizient mit Python und wichtigen Bibliotheken zu verarbeiten, zu sampeln und zu analysieren. Jede Sektion bietet anschauliche Videoerklärungen und interaktive Aufgaben, um Ihre Fachkenntnisse auszubauen.

Erlernen grundlegender Strategien zum Umgang mit Datensätzen, die zu groß sind, um in den Arbeitsspeicher zu passen, einschließlich Chunking- und Streaming-Techniken.

Untersuchung von Methoden zum Ausbalancieren und Sampeln großer Datensätze, einschließlich Oversampling und Undersampling.

Erlernen Sie die Verwendung der Polars-Bibliothek für schnelle, speichereffiziente Datenverarbeitung.

Challenge: Filtern Großer Datensätze

Lösung