En este desafío, utilizarás **polars** para realizar de manera eficiente la agregación de datos en conjuntos de datos grandes. Específicamente, se te pide agrupar un DataFrame grande por una columna y calcular la media de otra columna para cada grupo. Esta es una operación común en el análisis de datos, especialmente al trabajar con big data, ya que permite resumir y extraer información de subconjuntos de tus datos sin cargar todo en la memoria al mismo tiempo.

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

Un curso práctico y orientado a la acción para aspirantes a científicos de datos preparados para enfrentar desafíos reales con grandes volúmenes de datos. Aprenda a procesar, muestrear y analizar conjuntos de datos masivos de manera eficiente utilizando Python y bibliotecas esenciales. Cada sección incluye explicaciones en video atractivas y desafíos interactivos para desarrollar su experiencia.

Aprende estrategias fundamentales para manejar conjuntos de datos demasiado grandes para caber en la memoria, incluyendo técnicas de particionado y transmisión.

Explora métodos para equilibrar y muestrear grandes conjuntos de datos, incluyendo el sobremuestreo y el submuestreo.

Aprende a utilizar la biblioteca Polars para un procesamiento de datos rápido y eficiente en memoria.

Challenge: Agregación de Datos con Polars

Solución