Kuvittele, että tehtävänäsi on analysoida valtava CSV-tiedosto, joka sisältää miljoonia rivejä—liian suuri ladattavaksi muistiin kerralla. Tavoitteena on poimia vain ne rivit, joissa tietyn sarakkeen arvo ylittää annetun raja-arvon, ja tallentaa suodatetut tulokset uuteen tiedostoon. Tämä tilanne on yleinen suurten tietomäärien analysoinnissa, jossa tehokas ja muistia säästävä käsittely on olennaista.


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

Käytännönläheinen, hands-on-kurssi aloitteleville data-analyytikoille, jotka ovat valmiita kohtaamaan todellisen maailman suurten datamäärien haasteet. Opettele käsittelemään, otantamaan ja analysoimaan massiivisia tietoaineistoja tehokkaasti Pythonilla ja keskeisillä kirjastoilla. Jokainen osio sisältää mukaansatempaavia videoesityksiä ja interaktiivisia haasteita asiantuntemuksen kehittämiseksi.

Opi perustavanlaatuisia strategioita liian suurten tietoaineistojen käsittelyyn, jotka eivät mahdu muistiin, mukaan lukien paloittelu- ja suoratoistotekniikat.

Tutustu menetelmiin suurten tietoaineistojen tasapainottamiseksi ja näytteistämiseksi, mukaan lukien ylänäytteistäminen ja alinäytteistäminen.

Opi käyttämään polars-kirjastoa nopeaan ja muistitehokkaaseen datankäsittelyyn.

Challenge: Suurten Tietoaineistojen Suodatus

Ratkaisu