Уявіть, що вам потрібно проаналізувати масивний CSV-файл, який містить мільйони записів — занадто великий, щоб завантажити його повністю в оперативну пам'ять. Ваша мета — витягти лише ті рядки, де значення певного стовпця перевищує заданий поріг, і зберегти відфільтровані результати у новий файл. Така ситуація є типовою для аналізу великих обсягів даних, де важлива ефективна обробка з мінімальним використанням пам'яті.


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

Практичний курс для майбутніх дата-сайентістів, які готові вирішувати реальні завдання з обробки великих обсягів даних. Навчіться ефективно обробляти, вибірково аналізувати та аналізувати масивні набори даних за допомогою Python та основних бібліотек. Кожен розділ містить захоплюючі відео-пояснення та інтерактивні завдання для розвитку ваших навичок.

Вивчення основних стратегій обробки наборів даних, які занадто великі для розміщення в оперативній пам'яті, включаючи методи розбиття на частини та потокової обробки.

Дослідження методів балансування та вибірки великих наборів даних, включаючи оверсемплінг та андерсемплінг.

Вивчення використання бібліотеки polars для швидкої та ефективної обробки даних з мінімальним використанням пам'яті.

Завдання: Фільтрація Великих Наборів Даних

Рішення