python - 在 MLFlow 管道中读取 csv 文件

Question

我正在关注 this documentation使用 ML Flow 管道，这需要克隆 this repository .如果我按原样运行完整的管道，它会完美运行:

import os
from mlflow.pipelines import Pipeline

os.chdir("~/mlp-regression-template")
regression_pipeline = Pipeline(profile="local")
# Display a visual overview of the pipeline graph
regression_pipeline.inspect()
# Run the full pipeline
regression_pipeline.run()

但是当我尝试更改第一部分以读取不同的数据集时，出现以下错误:

mlflow.exceptions.MlflowException: Resolved data file with path '/tmp/tmpv201mpms/precio_leche.csv' does not have the expected format 'parquet'.

这是正确的，输入文件现在不是 csv 格式，我在数据文件夹中添加了一个新文件并更改了配置文件 local.yaml 文件:

INGEST_DATA_LOCATION: "./data/precio_leche.csv" 

我不明白的是，在管道中的 insgest 步骤中，执行 ingest.py 代码:

将读取的csv文件转换成哪个函数:

def load_file_as_dataframe(file_path: str, file_format: str) -> DataFrame:
    """
    Load content from the specified dataset file as a Pandas DataFrame.

    This method is used to load dataset types that are not natively  managed by MLflow Pipelines
    (datasets that are not in Parquet, Delta Table, or Spark SQL Table format). This method is
    called once for each file in the dataset, and MLflow Pipelines automatically combines the
    resulting DataFrames together.

    :param file_path: The path to the dataset file.
    :param file_format: The file format string, such as "csv".
    :return: A Pandas DataFrame representing the content of the specified file.
    """

    if file_format == "csv":
        import pandas

        _logger.warning(
            "Loading dataset CSV using `pandas.read_csv()` with default arguments and assumed index"
            " column 0 which may not produce the desired schema. If the schema is not correct, you"
            " can adjust it by modifying the `load_file_as_dataframe()` function in"
            " `steps/ingest.py`"
        )
        return pandas.read_csv(file_path, index_col=0)
    else:
        raise NotImplementedError

出现此错误后，我还尝试更改 pipeline.yaml 文件以将 .csv 作为默认格式:

  format: {{INGEST_DATA_FORMAT|default('csv')}}

但它也没有用，我还注意到，当我为默认数据集运行摄取步骤时，它会从 pandas 分析库返回此摘要:

但我没有看到代码中的哪个位置，我是否更改了错误的文件？或者我应该怎么做才能在摄取步骤中读取 csv 文件？还。我希望阅读多个文件，而不仅仅是一个。

Answer 1

根据 MLflow 的 documentations ，您的 local.yaml 文件应如下所示:

data:
  location: {{INGEST_DATA_LOCATION|default('~/your/custom/data/path/single_file.csv')}}
  format: {{INGEST_DATA_FORMAT|default('csv')}}
  custom_loader_method: steps.ingest.load_file_as_dataframe

而不是您在问题中分享的内容。

对于多文件摄取，您可能希望将文件保存在 INGEST_DATA_LOCATION 和/或修改默认设置以包含通配符，例如 *.csv