Я пытаюсь настроить конвейер Sagemaker, который состоит из двух этапов: предварительная обработка, а затем обучение модели RF. На первом шаге создаются 3 вывода: scaled_data.csv, train.csv и test.csv. На втором этапе необходимо обучить и протестировать CSV для обучения RF-модели. При выполнении шага 2 возникает ошибка, в которой говорится: «Ошибка типа: объект типа «Свойства» не сериализуем JSON».
Вот мой код для настройки шагов конвейера:
# upload data from local path to default bucket with prefix raw_data
WORK_DIRECTORY = "data"
input_data = sagemaker_session.upload_data(
path = "{}/{}".format(WORK_DIRECTORY, "dataset.csv"),
bucket=bucket,
key_prefix = "{}/{}".format(prefix, "input_data"),
)
scaling_processor = SKLearnProcessor(
framework_version=FRAMEWORK_VERSION,
instance_type = "ml.m5.4xlarge",
instance_count=processing_instance_count,
base_job_name = "data-process",
role=role,
sagemaker_session=pipeline_session,
)
scaling_processor_args = scaling_processor.run(
inputs=[
ProcessingInput(source=input_data, destination = "/opt/ml/processing/input"),
],
outputs=[
ProcessingOutput(output_name = "scaled_data", source = "/opt/ml/processing/output/scaled_data/"),
ProcessingOutput(output_name = "train", source = "/opt/ml/processing/output/train/"),
ProcessingOutput(output_name = "test", source = "/opt/ml/processing/output/test/") ],
code = "scripts/preprocess.py",
)
step_process = ProcessingStep(name = "DataProcess", step_args=scaling_processor_args)
estimator_cls = sagemaker.sklearn.SKLearn
FRAMEWORK_VERSION = "0.23-1"
rf_processor = FrameworkProcessor(
estimator_cls,
FRAMEWORK_VERSION,
role = role,
instance_count=1,
instance_type='ml.m5.2xlarge',
base_job_name='rf-modelling'
)
rf_processor_args = rf_processor.run(
inputs=[
ProcessingInput(source=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
destination = "/opt/ml/processing/input"),
ProcessingInput(source=step_process.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
destination = "/opt/ml/processing/input"), ],
outputs=[
ProcessingOutput(output_name = "rf_model",source = "/opt/ml/processing/output/")
],
code = "scripts/train.py",
)
step_train = ProcessingStep(name = "RFTrain", step_args=rf_processor_args)
При выполнении шага 2 возникает ошибка, в которой говорится: «Ошибка типа: объект типа «Свойства» не сериализуем JSON». Проблема в строке, в которой я устанавливаю processingInput для 2-го шага в rf_processor_args. Любые идеи, что может вызвать эту ошибку?






run(), который вы вызываете, кажется неправильным выбором, поскольку run() используется для непосредственного запуска задания обработки вместо определения шагов конвейера, которые вы, по-видимому, хотите. Используйте ProcessingStep напрямую и наполните его всеми необходимыми аргументами:
# Setup the first step (scaling step)
...
# --> Use ProcessingStep directly and provide all the args
step_process = ProcessingStep(
name = "DataProcess",
processor=scaling_processor,
inputs=[
ProcessingInput(source=input_data, destination = "/opt/ml/processing/input"),
],
outputs=[
ProcessingOutput(output_name = "scaled_data", source = "/opt/ml/processing/output/scaled_data/"),
ProcessingOutput(output_name = "train", source = "/opt/ml/processing/output/train/"),
ProcessingOutput(output_name = "test", source = "/opt/ml/processing/output/test/"),
],
code = "scripts/preprocess.py",
)
# Setup the 2nd step (RF Training-BYO mode)
...
# --> Use ProcessingStep directly and provide all the args
step_train = ProcessingStep(
name = "RFTrain",
processor=rf_processor,
inputs=[
ProcessingInput(source=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
destination = "/opt/ml/processing/input/train"),
ProcessingInput(source=step_process.properties.ProcessingOutputConfig.Outputs["test"].S3Output.S3Uri,
destination = "/opt/ml/processing/input/test"),
],
outputs=[
ProcessingOutput(output_name = "rf_model", source = "/opt/ml/processing/output/"),
],
code = "scripts/train.py",
)