Блог

• Я пытаюсь прочитать файл из хранилища больших двоичных объектов, загрузить в pandas и записать его в BlobStorage

• У меня есть конвейер машинного обучения Azure с PythonScriptStep, который принимает 2 конвейерных параметра и представляет собой пути к данным, как показано ниже.

   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)
  

Запуск этого AzureMLPipeline из AzureDataFactory :
Приведенный выше код отлично работает, выполняя конвейер ML в AzureMLWorkspace/PipelineSDK. Я пытаюсь запустить действие AzureMLpipeline из AzureDataFactory (AzureMachineLearningExecutePipeline) следующим образом

Попробовал выполнить отладку следующим образом, передав 2 пути ввода строки
rawdatapath = «samplefolder/raw/alerts/2017/05/31/ test.csv»
cleaneddatapath = «samplefolder/ raw / очищенный/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'
  

Это показывает, что вместо параметра конвейера используется путь по умолчанию (Нет такой ошибки файла или каталога, которая менее важна, поскольку основной момент заключается в том, что вместо параметров конвейера используется путь по умолчанию). Я сомневаюсь, что это связано с передачей конвейерного параметра в виде строки вместо пути к данным.

НАКОНЕЦ, ВОПРОС: как передать datapath в azuremplpipelineactivity из Azure Data Factory?

Спасибо.

Эта записная книжка демонстрирует использование DataPath и PipelineParameters в конвейере AML. Вы узнаете, как строки и DataPath могут быть параметризованы и отправлены в конвейеры AML через PipelineParameters . Вы можете параметризовать входной набор данных, и вот пример блокнота, который показывает, как это сделать.

В настоящее время ParallelRunStep принимает dataset в качестве входных данных. вы можете добавить еще один шаг перед ParallelRunStep созданием объекта dataset, указывающего на новые данные, и передать в ParallelRunStep . Вот пример использования нескольких шагов:

Для вывода, если вы используете append_row действие вывода, вы можете настроить имя выходного файла через append_row_file_name config. Выходные данные будут сохранены в большом двоичном объекте по умолчанию. Чтобы переместить его в другое хранилище, мы предлагаем использовать другое DataTransferStep после ParallelRunStep . Пожалуйста, следуйте этому примеру для шага передачи данных:

Получил ответ от Microsoft (пожалуйста, обратитесь к этой теме здесь). Команда разработчиков Azure Data Factory подтверждает, что на сегодняшний день на фабрике данных Azure (ADF) не поддерживается тип данных для параметра «DataPath». Однако для этого уже есть функция, и над ней ведется работа. Эта функция будет частью ноябрьского выпуска.

Входные параметры, похоже, определены как строка, попробуйте изменить их как тип данных объекта. Согласно документации, он ожидает параметры объекта {«Ключ»: «значение»}.