Mage ist ein leistungsstarkes Tool für ETL-Aufgaben mit Funktionen, die die Datenexploration und -gewinnung, schnelle Visualisierungen durch Diagrammvorlagen und mehrere andere Funktionen ermöglichen, die Ihre Arbeit mit Daten in etwas Magisches verwandeln.

Bei der Datenverarbeitung kommt es während eines ETL-Prozesses häufig vor, dass fehlende Daten gefunden werden, die in der Zukunft zu Problemen führen können. Abhängig von der Aktivität, die wir mit dem Datensatz ausführen, können Nulldaten ziemlich störend sein.

Um das Fehlen von Daten in unserem Datensatz zu identifizieren, können wir Python und die Pandas-Bibliothek verwenden, um die Daten zu überprüfen, die Nullwerte darstellen. Darüber hinaus können wir Diagramme erstellen, die die Auswirkungen dieser Nullwerte noch deutlicher zeigen Unser Datensatz.

Unsere Pipeline besteht aus 4 Schritten: Beginnend mit dem Laden der Daten, zwei Verarbeitungsschritten und dem Exportieren der Daten.

Datenlader

In diesem Artikel verwenden wir den Datensatz: Binary Prediction of Poisonous Mushrooms, der im Rahmen eines Wettbewerbs auf Kaggle verfügbar ist. Nutzen wir den auf der Website verfügbaren Trainingsdatensatz.

Lassen Sie uns einen Data Loader-Schritt mit Python erstellen, um die Daten laden zu können, die wir verwenden werden. Vor diesem Schritt habe ich eine Tabelle in der Postgres-Datenbank erstellt, die ich lokal auf meinem Rechner habe, um die Daten laden zu können. Da die Daten in Postgres vorliegen, verwenden wir die bereits definierte Postgres-Ladevorlage in Mage.

from mage_ai.settings.repo import get_repo_path
from mage_ai.io.config import ConfigFileLoader
from mage_ai.io.postgres import Postgres
from os import path

if 'data_loader' not in globals():
    from mage_ai.data_preparation.decorators import data_loader

if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

@data_loader
def load_data_from_postgres(*args, **kwargs):
    """
    Template for loading data from a PostgreSQL database.
    Specify your configuration settings in 'io_config.yaml'.
    Docs: https://docs.mage.ai/design/data-loading#postgresql
    """
    query = 'SELECT * FROM mushroom'  # Specify your SQL query here
    config_path = path.join(get_repo_path(), 'io_config.yaml')
    config_profile = 'default'

    with Postgres.with_config(ConfigFileLoader(config_path, config_profile)) as loader:

        return loader.load(query)

@test
def test_output(output, *args) -> None:
    """
    Template code for testing the output of the block.
    """

    assert output is not None, 'The output is undefined'

Innerhalb der Funktion load_data_from_postgres() definieren wir die Abfrage, die wir zum Laden der Tabelle in die Datenbank verwenden. In meinem Fall habe ich die Bankinformationen in der Datei io_config.yaml konfiguriert, wo sie als Standardkonfiguration definiert ist, sodass wir nur den Standardnamen an die Variable config_profile übergeben müssen.

Nach der Ausführung des Blocks verwenden wir die Funktion „Diagramm hinzufügen“, die Informationen über unsere Daten über bereits definierte Vorlagen bereitstellt. Klicken Sie einfach auf das Symbol neben dem Play-Button, im Bild mit einer gelben Linie markiert.

Wir werden zwei Optionen auswählen, um unseren Datensatz weiter zu untersuchen: die Optionen summay_overview und feature_profiles. Durch summary_overview erhalten wir Informationen über die Anzahl der Spalten und Zeilen im Datensatz. Wir können auch die Gesamtzahl der Spalten nach Typ anzeigen, beispielsweise die Gesamtzahl der kategorialen, numerischen und booleschen Spalten. Feature_profiles hingegen stellt beschreibendere Informationen zu den Daten dar, wie zum Beispiel: Typ, Minimalwert, Maximalwert und andere Informationen. Wir können sogar die fehlenden Werte visualisieren, die im Mittelpunkt unserer Behandlung stehen.

Um uns stärker auf fehlende Daten konzentrieren zu können, verwenden wir die Vorlage: % der fehlenden Werte, ein Balkendiagramm mit dem Prozentsatz der fehlenden Daten in jeder Spalte.

Das Diagramm enthält 4 Spalten, in denen fehlende Werte mehr als 80 % des Inhalts ausmachen, und weitere Spalten, in denen fehlende Werte, jedoch in geringeren Mengen, angezeigt werden. Diese Informationen ermöglichen es uns nun, nach verschiedenen Strategien zu suchen, um damit umzugehen Nulldaten.

Transformer-Drop-Spalten

Für Spalten, die mehr als 80 % Nullwerte haben, besteht die Strategie, die wir verfolgen werden, darin, Spalten im Datenrahmen zu löschen und die Spalten auszuwählen, die wir aus dem Datenrahmen ausschließen möchten. Mithilfe des Blocks TRANSFORMER in der Python-Sprache wählen wir die Option Spaltenentfernung aus.

from mage_ai.data_cleaner.transformer_actions.base import BaseAction
from mage_ai.data_cleaner.transformer_actions.constants import ActionType, Axis
from mage_ai.data_cleaner.transformer_actions.utils import build_transformer_action
from pandas import DataFrame

if 'transformer' not in globals():
    from mage_ai.data_preparation.decorators import transformer

if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

@transformer
def execute_transformer_action(df: DataFrame, *args, **kwargs) -> DataFrame:
    """
    Execute Transformer Action: ActionType.REMOVE
    Docs: https://docs.mage.ai/guides/transformer-blocks#remove-columns
    """
    action = build_transformer_action(
        df,
        action_type=ActionType.REMOVE,
        arguments=['veil_type', 'spore_print_color', 'stem_root', 'veil_color'],        
        axis=Axis.COLUMN,
    )
    return BaseAction(action).execute(df)

@test
def test_output(output, *args) -> None:
    """
    Template code for testing the output of the block.

    """
    assert output is not None, 'The output is undefined'

Innerhalb der Funktion execute_transformer_action() fügen wir in der Argumentvariablen eine Liste mit den Namen der Spalten ein, die wir aus dem Datensatz ausschließen möchten. Nach diesem Schritt führen Sie einfach den Block aus.

Transformator ergänzt fehlende Werte

Nun verwenden wir für die Spalten, die weniger als 80 % Nullwerte haben, die Strategie Fehlende Werte ausfüllen, da wir in einigen Fällen trotz fehlender Daten diese durch Werte wie ersetzen Abhängig von Ihrem endgültigen Ziel kann es möglicherweise den Datenbedarf decken, ohne viele Änderungen am Datensatz zu verursachen.

Es gibt einige Aufgaben, wie z. B. die Klassifizierung, bei denen das Ersetzen fehlender Daten durch einen für den Datensatz relevanten Wert (Modus, Mittelwert, Median) zum Klassifizierungsalgorithmus beitragen kann, der zu anderen Schlussfolgerungen führen könnte, wenn die Daten gelöscht würden wie in der anderen Strategie, die wir verwendet haben.

Um eine Entscheidung darüber zu treffen, welche Messung wir verwenden werden, verwenden wir erneut die Funktion Diagramm hinzufügen von Mage. Mithilfe der Vorlage Häufigste Werte können wir den Modus und die Häufigkeit dieses Werts in jeder der Spalten visualisieren.

Wir folgen den Schritten, die den vorherigen ähneln, und verwenden den Transformator Füllen Sie fehlende Werte ein, um die Aufgabe des Subtrahierens der fehlenden Daten mithilfe des Modus jeder der Spalten auszuführen: Steam_surface, Gill_spacing, Cap_surface , gill_attachment, ring_type.

from mage_ai.data_cleaner.transformer_actions.constants import ImputationStrategy
from mage_ai.data_cleaner.transformer_actions.base import BaseAction
from mage_ai.data_cleaner.transformer_actions.constants import ActionType, Axis
from mage_ai.data_cleaner.transformer_actions.utils import build_transformer_action
from pandas import DataFrame

if 'transformer' not in globals():
    from mage_ai.data_preparation.decorators import transformer

if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

@transformer
def execute_transformer_action(df: DataFrame, *args, **kwargs) -> DataFrame:

    """
    Execute Transformer Action: ActionType.IMPUTE
    Docs: https://docs.mage.ai/guides/transformer-blocks#fill-in-missing-values

    """
    action = build_transformer_action(
        df,
        action_type=ActionType.IMPUTE,
        arguments=df.columns,  # Specify columns to impute
        axis=Axis.COLUMN,
        options={'strategy': ImputationStrategy.MODE},  # Specify imputation strategy
    )

    return BaseAction(action).execute(df)


@test
def test_output(output, *args) -> None:
    """
    Template code for testing the output of the block.
    """
    assert output is not None, 'The output is undefined'

In der Funktion execute_transformer_action() definieren wir die Strategie zum Ersetzen von Daten in einem Python-Wörterbuch. Für weitere Ersatzoptionen greifen Sie einfach auf die Transformer-Dokumentation zu: https://docs.mage.ai/guides/transformer-blocks#fill-in-missing-values.

Datenexporteur

Bei der Durchführung aller Transformationen speichern wir unseren jetzt behandelten Datensatz in derselben Postgres-Datenbank, jetzt jedoch unter einem anderen Namen, damit wir ihn unterscheiden können. Mit dem Block Data Exporter und der Auswahl von Postgres definieren wir das Shema und die Tabelle, in der wir speichern möchten. Dabei berücksichtigen wir, dass die Datenbankkonfigurationen zuvor in der Datei io_config.yaml gespeichert wurden.

from mage_ai.settings.repo import get_repo_path
from mage_ai.io.config import ConfigFileLoader
from mage_ai.io.postgres import Postgres
from pandas import DataFrame
from os import path

if 'data_exporter' not in globals():
    from mage_ai.data_preparation.decorators import data_exporter

@data_exporter
def export_data_to_postgres(df: DataFrame, **kwargs) -> None:

    """
    Template for exporting data to a PostgreSQL database.
    Specify your configuration settings in 'io_config.yaml'.
    Docs: https://docs.mage.ai/design/data-loading#postgresql

    """

    schema_name = 'public'  # Specify the name of the schema to export data to
    table_name = 'mushroom_clean'  # Specify the name of the table to export data to
    config_path = path.join(get_repo_path(), 'io_config.yaml')
    config_profile = 'default'

    with Postgres.with_config(ConfigFileLoader(config_path, config_profile)) as loader:

        loader.export(
            df,
            schema_name,
            table_name,
            index=False,  # Specifies whether to include index in exported table
            if_exists='replace', #Specify resolution policy if table name already exists
        )

Vielen Dank und bis zum nächsten Mal?

repo -> https://github.com/DeadPunnk/Mushrooms/tree/main