Ao trabalhar com grandes conjuntos de dados, muitas vezes é necessário realizar agregações sem carregar o arquivo inteiro na memória. Uma tarefa comum é somar os valores de uma coluna específica em um arquivo CSV muito grande. Como o arquivo pode não caber na memória, é possível processá-lo em partes gerenciáveis usando a função `pandas` do `read_csv()` com o parâmetro `chunksize`.

Para cada parte, calcula-se a soma da coluna desejada e, em seguida, agrega-se essas somas parciais para obter o total. Essa abordagem é eficiente e escalável, permitindo lidar com arquivos de praticamente qualquer tamanho, desde que cada parte caiba na memória.


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

class TestTask(unittest.TestCase):
    def setUp(self):
        # Create test CSV files for different scenarios
        self.one_chunk_file = "test_one_chunk.csv"
        self.multi_chunk_file = "test_multi_chunk.csv"
        self.float_file = "test_float.csv"
        self.empty_col_file = "test_empty_col.csv"
        self.no_col_file = "test_no_col.csv"

        # One chunk: 3 rows
        with open(self.one_chunk_file, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["amount", "desc"])
            writer.writerow([10, "a"])
            writer.writerow([20, "b"])
            writer.writerow([30, "c"])
        # Multi chunk: 6 rows
        with open(self.multi_chunk_file, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["value", "desc"])
            for i in range(1, 7):
                writer.writerow([i, f"row{i}"])
        # Float values
        with open(self.float_file, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["price", "desc"])
            writer.writerow([1.5, "x"])
            writer.writerow([2.5, "y"])
            writer.writerow([3.0, "z"])
        # Empty column
        with open(self.empty_col_file, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["col1", "col2"])
            writer.writerow(["", "a"])
            writer.writerow(["", "b"])
        # No such column
        with open(self.no_col_file, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["foo", "bar"])
            writer.writerow([1, 2])
            writer.writerow([3, 4])

    def tearDown(self):
        for f in [self.one_chunk_file, self.multi_chunk_file, self.float_file, self.empty_col_file, self.no_col_file]:
            if os.path.exists(f):
                os.remove(f)

    def test_sum_one_chunk(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.sum_column_in_chunks(self.one_chunk_file, "amount", 100)
        _dynamic_test(
            self,
            result == 60,
            "Returns correct sum when file fits in one chunk.",
            f"Expected 60, got {result} for one chunk file."
        )

    def test_sum_multiple_chunks(self):
        import user_code
        importlib.reload(user_code)
        # chunksize=2, so 3 chunks
        result = user_code.sum_column_in_chunks(self.multi_chunk_file, "value", 2)
        _dynamic_test(
            self,
            result == 21,
            "Returns correct sum when file is read in multiple chunks.",
            f"Expected 21, got {result} for multi-chunk file."
        )

    def test_sum_float_column(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.sum_column_in_chunks(self.float_file, "price", 2)
        _dynamic_test(
            self,
            abs(result - 7.0) < 1e-8,
            "Returns correct sum for float column.",
            f"Expected 7.0, got {result} for float column."
        )

    def test_sum_empty_column(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.sum_column_in_chunks(self.empty_col_file, "col1", 1)
        _dynamic_test(
            self,
            result == 0,
            "Returns 0 for column with no values.",
            f"Expected 0, got {result} for empty column."
        )

    def test_sum_column_not_exist(self):
        import user_code
        importlib.reload(user_code)
        result = user_code.sum_column_in_chunks(self.no_col_file, "notacol", 1)
        _dynamic_test(
            self,
            result == 0,
            "Returns 0 if column does not exist.",
            f"Expected 0, got {result} for missing column."
        )

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

Um curso prático e aplicado para aspirantes a cientistas de dados prontos para enfrentar desafios reais com grandes volumes de dados. Aprenda a processar, amostrar e analisar conjuntos de dados massivos de forma eficiente utilizando Python e bibliotecas essenciais. Cada seção apresenta explicações em vídeo envolventes e desafios interativos para desenvolver sua expertise.

Aprenda estratégias fundamentais para lidar com conjuntos de dados grandes demais para caber na memória, incluindo técnicas de divisão em blocos e streaming.

Explore métodos para balancear e amostrar grandes conjuntos de dados, incluindo oversampling e undersampling.

Aprenda a usar a biblioteca Polars para processamento de dados rápido e eficiente em memória.

Desafio: Agregação de Dados em Blocos

Solução