3

  • 我正在尝试从 Blob Storage 读取文件，加载到 pandas 并将其写入 BlobStorage

  • 我有一个带有 PythonScriptStep 的 Azure 机器学习管道，它采用 2 个 PipelineParameters 并且是如下的 DataPath。

    from azureml.core import Datastore
from azureml.data.datapath import DataPath, DataPathComputeBinding, DataReference
from azureml.pipeline.core import PipelineParameter

datastore = Datastore(ws, "SampleStore")
in_raw_path_default = 'somefolder/raw/alerts/2020/08/03/default_in.csv'
in_cleaned_path_default= 'somefolder/cleaned/alerts/2020/08/03/default_out.csv'

in_raw_datapath = DataPath(datastore=datastore, path_on_datastore=in_raw_path_default)
in_raw_path_pipelineparam = PipelineParameter(name="inrawpath", default_value=in_raw_datapath)
raw_datapath_input = (in_raw_path_pipelineparam, DataPathComputeBinding(mode='mount'))

in_cleaned_datapath = DataPath(datastore=datastore, path_on_datastore=in_cleaned_path_default)
in_cleaned_path_pipelineparam = PipelineParameter(name="incleanedpath", default_value=in_cleaned_datapath)
cleaned_datapath_input = (in_cleaned_path_pipelineparam, DataPathComputeBinding(mode='mount'))

from azureml.pipeline.steps import PythonScriptStep

source_directory = script_folder + '/pipeline_Steps'
dataprep_step = PythonScriptStep(
    script_name="SimpleTest.py", 
    arguments=["--input_data", raw_datapath_input, "--cleaned_data", cleaned_datapath_input],
    inputs=[raw_datapath_input, cleaned_datapath_input],    
    compute_target=default_compute, 
    source_directory=source_directory,
    runconfig=run_config,
    allow_reuse=True
)

from azureml.pipeline.core import Pipeline
pipeline_test = Pipeline(workspace=ws, steps=[dataprep_step])

test_raw_path = DataPath(datastore=datastore, path_on_datastore='samplefolder/raw/alerts/2017/05/31/test.csv')
test_cleaned_path = DataPath(datastore=datastore, path_on_datastore='samplefolder/cleaned/alerts/2020/09/03')
pipeline_run_msalerts = Experiment(ws, 'SampleExperiment').submit(pipeline_test, pipeline_parameters={"inrawpath"  : test_raw_path,
                                                                                                        "incleanedpath" : test_cleaned_path})```

这是使用的脚本（SimpleTest.py）：

import os
import sys
import argparse
import pathlib
import azureml.core
import pandas as pd

parser = argparse.ArgumentParser("datapreponly")
parser.add_argument("--input_data", type=str)
parser.add_argument("--cleaned_data", type=str)

args = parser.parse_args()

print("Argument 1: %s" % args.input_data)
print("Argument 2: %s" % args.cleaned_data)

testDf = pd.read_csv(args.input_data, error_bad_lines=False)
print('Total Data Shape' + str(testDf.shape))

if not (args.cleaned_data is None):
    output_path = args.cleaned_data
    os.makedirs(output_path, exist_ok=True)
    outdatapath = output_path + '/alert.csv'    
    testDf.to_csv(outdatapath, index=False)

从 AzureDataFactory 触发此 AzureMLipeline：
通过在 AzureMLWorkspace/PipelineSDK 中执行 ML 管道，上述代码可以正常工作。我正在尝试从 AzureDataFactory(AzureMachineLearningExecutePipeline) 活动触发 AzureMLpipeline，如下所示

在此处输入图像描述


通过传递 2 个字符串输入路径rawdatapath = "samplefolder/raw/alerts/2017/05/31/test.csv"
cleandatapath = "samplefolder/raw/cleaned/2020/09/03/"尝试调试运行如下

在此处输入图像描述

Current directory:  /mnt/batch/tasks/shared/LS_root/jobs/myazuremlworkspace/azureml/d8ee11ea-5838-46e5-a8ce-da2fbff5aade/mounts/workspaceblobstore/azureml/d8ee11ea-5838-46e5-a8ce-da2fbff5aade
Preparing to call script [ SimpleTest.py ] 
with arguments:
 ['--input_data', '/mnt/batch/tasks/shared/LS_root/jobs/myazuremlworkspace/azureml/d8ee11ea-5838-46e5-a8ce-da2fbff5aade/mounts/SampleStore/somefolder/raw/alerts/2020/08/03/default_in.csv',
 '--cleaned_data', '/mnt/batch/tasks/shared/LS_root/jobs/myazuremlworkspace/azureml/d8ee11ea-5838-46e5-a8ce-da2fbff5aade/mounts/SampleStore/somefolder/cleaned/alerts/2020/08/03/default_out.csv']
After variable expansion, calling script [ SimpleTest.py ] with arguments:
 ['--input_data', '/mnt/batch/tasks/shared/LS_root/jobs/myazuremlworkspace/azureml/d8ee11ea-5838-46e5-a8ce-da2fbff5aade/mounts/SampleStore/somefolder/raw/alerts/2020/08/03/default_in.csv',
 '--cleaned_data', '/mnt/batch/tasks/shared/LS_root/jobs/myazuremlworkspace/azureml/d8ee11ea-5838-46e5-a8ce-da2fbff5aade/mounts/SampleStore/somefolder/cleaned/alerts/2020/08/03/default_out.csv']

Script type = None
Argument 1: /mnt/batch/tasks/shared/LS_root/jobs/myazuremlworkspace/azureml/d8ee11ea-5838-46e5-a8ce-da2fbff5aade/mounts/SampleStore/somefolder/raw/alerts/2020/08/03/default_in.csv
Argument 2: /mnt/batch/tasks/shared/LS_root/jobs/myazuremlworkspace/azureml/d8ee11ea-5838-46e5-a8ce-da2fbff5aade/mounts/SampleStore/somefolder/cleaned/alerts/2020/08/03/default_out.csv
.......................
FileNotFoundError: [Errno 2] No such file or directory: '/mnt/batch/tasks/shared/LS_root/jobs/myazuremlworkspace/azureml/d8ee11ea-5838-46e5-a8ce-da2fbff5aade/mounts/SampleStore/somefolder/raw/alerts/2020/08/03/default_in.csv'

它表明采用默认路径而不是管道参数（没有这样的文件或目录错误不太重要，因为要点是采用默认路径而不是管道参数）。我怀疑它是因为将管道参数作为字符串而不是数据路径传递。


最后的问题：如何将数据路径从 Azure 数据工厂传递给 AzureMLipelineActivity？


谢谢。

4

3 回答 3

2

得到了微软的回答（请在此处参考此线程）。Azure 数据工厂产品团队确认，目前 Azure 数据工厂 (ADF) 中的“DataPath”参数不支持数据类型。但是，已经为此提出了一项功能，并且正在为此进行工作。此功能将成为 11 月版本的一部分。

于 2020-09-30T10:34:25.203 回答
2

这个笔记本演示了AML Pipeline的使用DataPath和在其中。PipelineParameters您将了解如何将字符串DataPath参数化并通过PipelineParameters. 您可以参数化输入数据集，这里是演示如何执行此操作的示例笔记本。

目前，ParallelRunStep接受数据集作为数据输入。您可以在ParallelRunStep创建指向新数据的数据集对象并传递给ParallelRunStep. 这是使用多个步骤的示例：

对于输出，如果使用append_row输出动作，可以通过append_row_file_nameconfig.xml 自定义输出文件名。输出将存储在默认 blob 中。要将其移动到其他商店，我们建议在DataTransferStep之后使用另一个ParallelRunStep。请按照此示例进行数据传输步骤：

于 2020-09-29T15:11:04.490 回答
1

输入参数似乎被定义为字符串，请尝试将它们修改为对象数据类型。根据文档，它需要 object {"Key" : "value"} 参数。

在此处输入图像描述

于 2020-09-16T00:15:25.230 回答