Mage 是用于 ETL 任务的强大工具，具有支持数据探索和挖掘的功能、通过图形模板进行快速可视化以及其他一些功能，可将您的数据工作转变为神奇的东西。

在数据处理中，在 ETL 过程中，经常会发现丢失的数据，这些数据可能会在将来产生问题，具体取决于我们要对数据集执行的活动，空数据可能会造成相当大的破坏。

为了识别数据集中是否缺少数据，我们可以使用 Python 和 pandas 库来检查呈现空值的数据，此外，我们可以创建图表来更清楚地显示这些空值的影响我们的数据集。

我们的管道由 4 个步骤组成：从加载数据开始，两个处理步骤和导出数据。

数据加载器

在本文中，我们将使用数据集：有毒蘑菇的二进制预测，该数据集可在 Kaggle 上作为竞赛的一部分获得。让我们使用网站上提供的训练数据集。

让我们使用 python 创建一个数据加载器步骤，以便能够加载我们将要使用的数据。在此步骤之前，我在 Postgres 数据库中创建了一个表，该数据库位于我的本地计算机上，以便能够加载数据。由于数据位于 Postgres 中，我们将使用 Mage 中已定义的 Postgres 加载模板。

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'

在函数load_data_from_postgres()中，我们将定义用于加载数据库中的表的查询。就我而言，我在文件 io_config.yaml 中配置了银行信息，其中它被定义为默认配置，因此我们只需将默认名称传递给变量 config_profile。

执行该块后，我们将使用“添加图表”功能，该功能将通过已定义的模板提供有关我们的数据的信息。只需单击播放按钮旁边的图标（在图像中用黄线标记）。

我们将选择两个选项来进一步探索我们的数据集，即 summay_overview 和 feature_profiles 选项。通过summary_overview，我们可以获得数据集中的列数和行数信息，我们还可以按类型查看列的总数，例如分类列、数字列和布尔列的总数。而Feature_profiles则呈现了更多关于数据的描述性信息，例如：类型、最小值、最大值等信息，我们甚至可以将缺失值可视化，这是我们处理的重点。

为了能够更多地关注缺失数据，让我们使用模板：缺失值百分比，这是一个条形图，每列中包含缺失数据的百分比。

该图显示了 4 列，其中缺失值对应于其内容的 80% 以上，而其他列则呈现缺失值，但数量较少，这些信息现在允许我们寻求不同的策略来处理此问题空数据。

变压器跌落柱

对于空值超过 80% 的列，我们将遵循的策略是在数据框中执行删除列，选择我们要从数据框中排除的列。使用Python语言中的

TRANSFORMER块，我们将选择选项列删除。

从 mage_ai.data_cleaner.transformer_actions.base 导入 BaseAction 从 mage_ai.data_cleaner.transformer_actions.constants 导入 ActionType，轴 从 mage_ai.data_cleaner.transformer_actions.utils 导入 build_transformer_action 从 pandas 导入 DataFrame 如果“transformer”不在 globals() 中：     从 mage_ai.data_preparation.decorators 导入变压器 如果“test”不在 globals() 中：     从 mage_ai.data_preparation.decorators 导入测试 @变压器 defexecute_transformer_action(df: DataFrame, *args, **kwargs) -> DataFrame:     ”“”     运行变压器操作：ActionType.REMOVE     文档：https://docs.mage.ai/guides/transformer-blocks#remove-columns     ”“”     动作 = build_transformer_action(         df,         action_type=ActionType.REMOVE,         参数=['veil_type'，'spore_print_color'，'stem_root'，'veil_color']，           轴=轴.COLUMN,     ）     返回 BaseAction(action).execute(df) @测试 def test_output(output, *args) -> 无：     ”“”     用于测试块输出的模板代码。     ”“”     断言输出不是 None，“输出未定义”

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'

在函数

execute_transformer_action()中，我们将在参数变量中插入一个列表，其中包含我们想要从数据集中排除的列的名称，在此步骤之后，只需执行该块即可。

变压器填充缺失值

现在，对于空值少于 80% 的列，我们将使用策略

填充缺失值，在某些情况下，尽管存在缺失数据，但仍将其替换为诸如以下的值平均，或时尚，它可能能够满足数据需求，而不会对数据集造成太多更改，具体取决于您的最终目标。

有一些任务，例如分类，用与数据集相关的值（众数、平均值、中位数）替换缺失数据，可以对分类算法做出贡献，如果删除数据，分类算法可以得出其他结论正如我们使用的其他策略一样。

为了决定我们将使用哪种测量，我们将再次使用 Mage 的

添加图表功能。使用模板最常见的值我们可以可视化每列中该值的众数和频率。

接下来的步骤与之前的类似，我们将使用变压器

填充缺失值，使用每列的模式来执行减去缺失数据的任务：steam_surface、gill_spacing、cap_surface 、gill_attachment、ring_type。

从 mage_ai.data_cleaner.transformer_actions.constants 导入 ImputationStrategy 从 mage_ai.data_cleaner.transformer_actions.base 导入 BaseAction 从 mage_ai.data_cleaner.transformer_actions.constants 导入 ActionType，轴 从 mage_ai.data_cleaner.transformer_actions.utils 导入 build_transformer_action 从 pandas 导入 DataFrame 如果“transformer”不在 globals() 中：     从 mage_ai.data_preparation.decorators 导入变压器 如果“test”不在 globals() 中：     从 mage_ai.data_preparation.decorators 导入测试 @变压器 defexecute_transformer_action(df: DataFrame, *args, **kwargs) -> DataFrame:     ”“”     运行转换器操作：ActionType.IMPUTE     文档：https://docs.mage.ai/guides/transformer-blocks#fill-in-missing-values     ”“”     动作 = build_transformer_action(         df,         action_type=ActionType.IMPUTE,         argument=df.columns, # 指定要插补的列         轴=轴.COLUMN,         options={'strategy': ImputationStrategy.MODE}, # 指定插补策略     ）     返回 BaseAction(action).execute(df) @测试 def test_output(output, *args) -> 无：     ”“”     用于测试块输出的模板代码。     ”“”     断言输出不是 None，“输出未定义”

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'

在函数

execute_transformer_action()中，我们定义了替换Python字典中数据的策略。如需更多替换选项，只需访问变压器文档：https://docs.mage.ai/guides/transformer-blocks#fill-in-missing-values。

数据导出器

执行所有转换时，我们将把现在处理的数据集保存在同一个 Postgres 数据库中，但现在使用不同的名称，以便我们可以区分。 使用

Data Exporter 块并选择 Postgres，我们将定义要保存的模式和表，记住数据库配置之前保存在文件 io_config.yaml 中。

从 mage_ai.settings.repo 导入 get_repo_path 从 mage_ai.io.config 导入 ConfigFileLoader 从 mage_ai.io.postgres 导入 Postgres 从 pandas 导入 DataFrame 从导入路径 如果“data_exporter”不在 globals() 中：     从 mage_ai.data_preparation.decorators 导入 data_exporter @data_exporter def export_data_to_postgres(df: DataFrame, **kwargs) -> 无：     ”“”     用于将数据导出到 PostgreSQL 数据库的模板。     在“io_config.yaml”中指定您的配置设置。     文档：https://docs.mage.ai/design/data-loading#postgresql     ”“”     schema_name = 'public' # 指定要将数据导出到的架构的名称     table_name = 'mushroom_clean' # 指定要将数据导出到的表的名称     config_path = path.join(get_repo_path(), 'io_config.yaml')     config_profile = '默认'     使用 Postgres.with_config(ConfigFileLoader(config_path, config_profile)) 作为加载程序：         加载器.导出（             df,             模式名称，             表名，             index=False, # 指定导出表中是否包含索引             if_exists='replace', #指定表名已存在时的解析策略         ）

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'

谢谢您，下次再见？

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