跨 Sagemaker 管道步骤共享代码，无需

我正在尝试通过多个步骤创建 Sagemaker 管道。我有一些代码想在不同的步骤中分享。下一个示例不是精确的，而是用于说明的简化版本。

我有如下所示的文件夹结构：

source_scripts/
├── utils
│   ├── logger.py
├── models/
│   ├── ground_truth.py
│   ├── document.py
├── processing/
│   ├── processing.py
│   └── main.py
└── training/
    ├── training.py
    └── main.py

我想使用 models 和 utils 内部 training.py 的代码，因为我不知道在哪里我正在使用的 Sagemaker 实例上安装的代码：

from ..common.ground_truth import GroundTruthRow

在构建管道时，我创建处理和训练步骤：

script_processor = FrameworkProcessor()
args = script_processor.get_run_args(
    source_dir="source_scripts"
    code="processing/main.py"
)
step_process = ProcessingStep(
    code=args.code
)

estimator = Estimator(
    source_dir="source_scripts"
    code="training/main.py"
)
step_train = TrainingStep(
   estimator=estimator
)

但在管道执行期间会导致错误：

ImportError: attempted relative import with no known parent package

有关如何在多个 SageMaker 作业之间共享代码的任何建议在单个管道中无需构建自定义 Docker 映像？

可以通过几种方法在 SageMaker 管道步骤之间共享代码，而无需构建自定义 Docker 映像：

1. 使用公共 S3 位置：

• 上传代码： 将的 utils 和 models 目录上传到 S3 中的一个公共存储桶。
• 在步骤中下载： 在处理和训练步骤的开头，从 S3 下载这些目录。
• 更新路径： 使用 sys.path.append() 将下载目录添加到 Python 路径中，以便可以导入模块。

```python import sys import os import boto3

# 在 S3 中存储代码的存储桶和前缀 bucket_name = 'your-bucket-name' prefix = 'your-code-prefix/'

# 下载代码 s3 = boto3.client('s3') for dir_name in ['utils', 'models']: local_dir = f'/tmp/{dir_name}' os.makedirs(local_dir, exist_ok=True) for obj in s3.list_objects_v2(Bucket=bucket_name, Prefix=f'{prefix}{dir_name}/')['Contents']: s3.download_file(bucket_name, obj['Key'], os.path.join(local_dir, os.path.basename(obj['Key'])))

# 更新 Python 路径 sys.path.append('/tmp/utils') sys.path.append('/tmp/models')

# 现在可以导入模块了 from utils.logger import logger from models.ground_truth import GroundTruthRow ```

2. 将代码作为依赖项包含在内：

• 创建 requirements.txt： 在 source_scripts 目录中创建一个 requirements.txt 文件，其中列出了的代码所需的所有依赖项，包括 utils 和 models 目录。
• 在 Estimator 和 Processor 中指定依赖项： 在创建 Estimator 和 FrameworkProcessor 时，使用 dependencies 参数指定 requirements.txt 文件的路径。这将确保在构建环境时安装依赖项。

```python # requirements.txt utils models other-dependencies

# 在 Estimator 和 Processor 中指定依赖项 script_processor = FrameworkProcessor( # ... other arguments ... dependencies=['source_scripts/requirements.txt'] )

estimator = Estimator( # ... other arguments ... dependencies=['source_scripts/requirements.txt'] ) ```

3. 使用代码目录作为源目录：

• 修改源目录： 将 source_dir 参数设置为 source_scripts 目录的父目录，该目录包含 utils 、 models 、 processing 和 training 。
• 更新导入： 更新的导入语句以包含完整的包路径。

```python # 修改后的目录结构 parent_directory/ └── source_scripts/ ├── utils/ ├── models/ ├── processing/ └── training/

# 更新后的导入 from source_scripts.utils.logger import logger from source_scripts.models.ground_truth import GroundTruthRow

# 创建 Estimator 和 Processor script_processor = FrameworkProcessor( source_dir='parent_directory', code='source_scripts/processing/main.py' )

estimator = Estimator( source_dir='parent_directory', code='source_scripts/training/main.py' ) ```

这些方法允许在 SageMaker 管道步骤之间有效共享代码，而无需创建自定义 Docker 映像，从而简化代码管理和部署。选择最适合需求和项目结构的方法。