Supposons que vous deviez analyser un fichier CSV volumineux contenant des millions d'enregistrements—trop important pour être chargé en mémoire en une seule fois. L'objectif est d'extraire uniquement les lignes où la valeur d'une colonne spécifique dépasse un certain seuil, puis d'enregistrer les résultats filtrés dans un nouveau fichier. Ce scénario est courant dans l'analyse de données à grande échelle, où un traitement efficace et économe en mémoire est essentiel.


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

Un cours pratique et appliqué destiné aux futurs data scientists prêts à relever les défis des grands volumes de données réels. Apprenez à traiter, échantillonner et analyser efficacement des ensembles de données massifs en utilisant Python et des bibliothèques essentielles. Chaque section propose des explications vidéo engageantes et des défis interactifs pour développer votre expertise.

Apprenez les stratégies fondamentales pour gérer des ensembles de données trop volumineux pour tenir en mémoire, y compris les techniques de découpage en blocs et de streaming.

Explorer des méthodes pour équilibrer et échantillonner de grands ensembles de données, y compris le suréchantillonnage et le sous-échantillonnage.

Apprenez à utiliser la bibliothèque Polars pour un traitement de données rapide et économe en mémoire.

Challenge: Filtrer de Grands Ensembles de Données

Solution