In dieser Aufgabe wird **polars** verwendet, um Datenaggregation effizient auf großen Datensätzen durchzuführen. Konkret besteht die Aufgabe darin, ein großes DataFrame nach einer Spalte zu gruppieren und für jede Gruppe den Mittelwert einer anderen Spalte zu berechnen. Dies ist ein gängiger Vorgang in der Datenanalyse, insbesondere bei der Arbeit mit Big Data, da so Teilmengen der Daten zusammengefasst und Erkenntnisse gewonnen werden können, ohne dass der gesamte Datensatz gleichzeitig in den Speicher geladen werden muss.

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

Ein praxisorientierter Kurs für angehende Data Scientists, die bereit sind, reale Herausforderungen mit großen Datenmengen zu meistern. Lernen Sie, massive Datensätze effizient mit Python und wichtigen Bibliotheken zu verarbeiten, zu sampeln und zu analysieren. Jede Sektion bietet anschauliche Videoerklärungen und interaktive Aufgaben, um Ihre Fachkenntnisse auszubauen.

Erlernen grundlegender Strategien zum Umgang mit Datensätzen, die zu groß sind, um in den Arbeitsspeicher zu passen, einschließlich Chunking- und Streaming-Techniken.

Untersuchung von Methoden zum Ausbalancieren und Sampeln großer Datensätze, einschließlich Oversampling und Undersampling.

Erlernen Sie die Verwendung der Polars-Bibliothek für schnelle, speichereffiziente Datenverarbeitung.

Herausforderung: Polars-Datenaggregation

Lösung