Python SageMaker Pipeline：类型错误：join() 参数必须是 str、bytes 或 os.PathLike 对象，而不是ParameterString

这里的问题是，您正在try 将一个Python字符串与一个类型为ParameterString的对象连接起来.os.path.join要求所有参数都是有效的路径类型(如字符串或字节)，这就是当您传递ParameterString时出现错误的原因.

在SageMaker管道中，ParameterString是一个占位符，表示管道的运行时参数.此参数的实际值仅在执行管道时可用，而不是在定义管道时可用.

要使用预处理脚本文件名对SageMaker管道进行参数化，您需要使用内置的SageMaker管道模板功能，该功能利用了Jinsa2模板引擎.

以下是您可以如何做到这一点:

1. 在code参数sklearn_processor.run()中使用{{ }}表示法可包含动态值data_preprocessing_script.

2. 按如下方式更新code参数:

code="{{ BASE_DIR }}/{{ data_preprocessing_script }}"

3. 将输入设置为step_args变量时，请使用模板化字符串:

step_args = sklearn_processor.run(
    code="{{ BASE_DIR }}/{{ data_preprocessing_script }}",
    inputs=[
        ProcessingInput(input_name="raw_input_data",
                        ...
    ]
)

4. 稍后，当您执行管道或创建管道执行时，可以提供data_preprocessing_script的实际值:

pipeline.execute(parameters={
    "DataPreProcessingScript": "preprocessing/xxx_preprocessing.py"
})

通过使用这种方法，您可以使您的代码更加动态，允许您在运行时指定预处理脚本，而不必每次都修改管道定义.

错误消息指示在预期位置找不到由data_preprocessing_script变量指定的文件.

需要注意的关键细节是:

1. Directory Base:

   您似乎正在使用BASE_DIR作为您的脚本所在的基本目录.确保该变量正确指向您的脚本所在的位置.

2. Jinja Templating:

   您正在使用JJJA模板({{ data_preprocessing_script }}).这意味着您希望使用某种模板机制将该占位符替换为实际的值.如果模板机制没有替换占位符，那么字符串{{ data_preprocessing_script }}将按原样使用，这可能会导致您看到的错误.

3. File Existence Check:

   在使用该文件之前，您可能需要判断它是否存在.这有助于调试:

   import os
   
   script_path = os.path.join(BASE_DIR, "{{ data_preprocessing_script }}")
   if not os.path.exists(script_path):
       raise ValueError(f"Script not found at {script_path}")
   

Example Code:

AWS SageMaker管道定义通常涉及定义各种步骤及其相关参数.在没有完整的管道定义上下文的情况下，很难提供准确的示例.

然而，这里有一个可能会有所帮助的通用示例:

import os
from sagemaker.workflow.steps import ProcessingStep
from sagemaker.processing import ScriptProcessor

BASE_DIR = "/home/ec2-user/SageMaker/pipelines/project"
DATA_PREPROCESSING_SCRIPT = "your_script_name.py"  # Replace with your script's name

script_path = os.path.join(BASE_DIR, DATA_PREPROCESSING_SCRIPT)
if not os.path.exists(script_path):
    raise ValueError(f"Script not found at {script_path}")

# Define the script processor
script_processor = ScriptProcessor(
    image_uri="your_container_image_uri",
    command=["python3"],
    instance_type="ml.m5.xlarge",
    instance_count=1,
    role="your_sagemaker_execution_role_arn"
)

# Define the processing step
step_process = ProcessingStep(
    name="DataPreprocessing",
    processor=script_processor,
    inputs=[...],  # Define your inputs
    outputs=[...], # Define your outputs
    code=script_path
)