Neste desafio, você utilizará o **polars** para realizar agregação de dados de forma eficiente em conjuntos de dados grandes. Especificamente, a tarefa é agrupar um DataFrame grande por uma coluna e calcular a média de outra coluna para cada grupo. Esta é uma operação comum em análise de dados, especialmente ao trabalhar com big data, pois permite resumir e extrair insights de subconjuntos dos seus dados sem carregar tudo na memória de uma só vez.

import unittest
import sys
import subprocess
import importlib
import ast
import re
import polars as pl

class TestTask(unittest.TestCase):
    def setUp(self):
        self.module = importlib.import_module("user_code")

    def test_returns_dataframe_with_one_row_per_group(self):
        df = pl.DataFrame({
            "group": ["X", "Y", "X", "Z", "Y", "Z", "X"],
            "score": [5, 10, 15, 20, 30, 40, 25]
        })
        result = self.module.groupby_mean(df, "group", "score")
        _dynamic_test(
            self,
            isinstance(result, pl.DataFrame),
            "Returned object is a polars DataFrame",
            f"Returned object is not a polars DataFrame: {type(result)}"
        )
        unique_groups = set(df["group"].to_list())
        _dynamic_test(
            self,
            result.height == len(unique_groups),
            "Returned DataFrame has one row per unique group value",
            f"Expected {len(unique_groups)} groups, got {result.height} rows"
        )

    def test_correct_mean_computation(self):
        df = pl.DataFrame({
            "team": ["A", "B", "A", "B", "A"],
            "points": [10, 20, 30, 40, 50]
        })
        result = self.module.groupby_mean(df, "team", "points")
        expected = {
            "A": (10 + 30 + 50) / 3,
            "B": (20 + 40) / 2
        }
        for row in result.iter_rows(named=True):
            group = row["team"]
            mean_val = row.get("mean_points")
            _dynamic_test(
                self,
                group in expected and abs(mean_val - expected[group]) < 1e-6,
                f"Mean for group {group} is correct",
                f"Expected mean {expected[group]} for group {group}, got {mean_val}"
            )

    def test_column_naming(self):
        df = pl.DataFrame({
            "cat": ["a", "b", "a"],
            "val": [1, 2, 3]
        })
        result = self.module.groupby_mean(df, "cat", "val")
        _dynamic_test(
            self,
            "mean_val" in result.columns,
            "Returned DataFrame column is named correctly as 'mean_val'",
            f"Returned DataFrame columns: {result.columns}"
        )

    def test_works_with_any_dataframe(self):
        df = pl.DataFrame({
            "city": ["NY", "LA", "NY", "SF"],
            "temp": [70, 80, 60, 65],
            "humidity": [30, 40, 50, 60]
        })
        # Test grouping by city, mean of temp
        result = self.module.groupby_mean(df, "city", "temp")
        expected = {
            "NY": (70 + 60) / 2,
            "LA": 80,
            "SF": 65
        }
        for row in result.iter_rows(named=True):
            group = row["city"]
            mean_val = row.get("mean_temp")
            _dynamic_test(
                self,
                group in expected and abs(mean_val - expected[group]) < 1e-6,
                f"Mean for city {group} is correct",
                f"Expected mean {expected[group]} for city {group}, got {mean_val}"
            )

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 com Polars

Solução