AWS SageMaker处理作业报错排查：ProcessingOutput配置异常

问题背景 #

按照SageMaker文档配置处理作业，用part_rel_processing.ipynb触发作业并管理路径，processing_participant_relationships.py执行数据预处理，但作业持续失败，报错指向ProcessingOutput(output_name="test_data", source="/opt/ml/processing/test")配置，不确定是output_name还是路径问题。

Notebook代码（part_rel_processing.ipynb） #

Block 1: 初始化处理器

import boto3
import sagemaker
from sagemaker import get_execution_role
from sagemaker.sklearn.processing import SKLearnProcessor

region = boto3.session.Session().region_name

role = get_execution_role()
sklearn_processor = SKLearnProcessor(
    framework_version="0.20.0", role=role, instance_type="ml.m5.xlarge", instance_count=1
)

Block 2: 运行处理作业（报错位置）

from sagemaker.processing import ProcessingInput, ProcessingOutput

sklearn_processor.run(
    code="processing_participant_relationships.py",
    inputs=[
        ProcessingInput(source="s3://sagemaker-us-east-1-898900188658/gun_violence_data", destination="/opt/ml/processing/input"),
    ],
    outputs=[
        ProcessingOutput(output_name="test_data", source="/opt/ml/processing/test"), # <---- 报错指向此处
    ],
)

Block 3: 获取输出路径

preprocessing_job_description = sklearn_processor.jobs[-1].describe()

output_config = preprocessing_job_description["ProcessingOutputConfig"]
for output in output_config["Outputs"]:
    if output["OutputName"] == "test_data":
        preprocessed_training_data = output["S3Output"]["S3Uri"]

预处理脚本代码（processing_participant_relationships.py） #

import argparse
import os
import warnings

import pandas as pd
import numpy as np
from sklearn.preprocessing import PolynomialFeatures
from sklearn.compose import make_column_transformer
from sklearn.exceptions import DataConversionWarning

warnings.filterwarnings(action="ignore", category=DataConversionWarning)


if __name__ == "__main__":
   
    input_data_path = os.path.join("'/opt/ml/processing/input", "gun_violence.csv")
    input_data_path = 's3://sagemaker-us-east-1-8989001XXXXX/gun_violence_data/gun-violence.csv' # 隐去桶ID

    print("Reading input data from {}".format(input_data_path))
    df = pd.read_csv(input_data_path)
   
    df['suspect_rel'] = ''

    # 解析各列格式
    for i, row in df.iterrows():

        temp = row['participant_type']
        #print("participant_type row %s" % temp)

        if isinstance(temp, float):
            continue
        #match = re.findall('\\d*::\\d*Subject-Suspect', temp)

        # 获取嫌疑人索引
        match = re.findall('\\d*::Subject-Suspect', temp)

        if len(match) == 0:
            continue
        elif 'Subject-Suspect' not in match[0]:
            continue


        for keyval in match:
            if '::' in str(keyval):
                #print("keyval: %s" % keyval)
                part_value = str(keyval).split('::')
                part_index = part_value[0]

                temp_age_group = row['participant_relationship']
                if isinstance(row['participant_relationship'], float):
                    pass
                else:
                    regex = part_index + '::(.*)'
                    #print("regex: %s" % regex)
                    #print("temp_age_group: %s" % temp_age_group)
                    if not isinstance(temp_age_group, float):
                        match_age = re.findall(regex, temp_age_group)
                        #print("match_age: %s" % match_age)
                        if len(match_age) != 0:
                            if '||' in match_age[0]:
                                element = match_age[0].split('||')
                                if element[0] == '':
                                    pass
                                    #print("empty element: --%s--" % element[0])
                                else:
                                    df.at[i, 'suspect_rel'] = element
                            else:
                                if match_age[0] == '':
                                    #print("do nothing")
                                    pass
                                else:
                                    df.at[i, 'suspect_rel'] = match_age
                                #print("i = %d" % i)
                        else:
                            continue

        print("Preprocessing Suspect Relationship CSV: {}".format(df.shape))

        parsed_suspect_relationship_output_path = os.path.join("/opt/ml/processing/participant_relationship", "parsed_suspect_relationship.csv")

报错信息（翻译后） #

处理作业失败，原因：处理容器退出时返回非零状态码1。检查作业日志获取详细信息。日志路径：s3://sagemaker-us-east-1-898900188658/sagemaker-scikit-learn-202X-XX-XX-XX-XX-XX-XXX/output/logs/jobs/sagemaker-scikit-learn-202X-XX-XX-XX-XX-XX-XXX/

问题根因与解决方案 #

1. 输出路径不匹配 #

Notebook配置的输出源路径是/opt/ml/processing/test，但预处理脚本实际生成的输出路径是/opt/ml/processing/participant_relationship，SageMaker找不到指定的输出目录，导致作业失败。

修复：统一输出路径，二选一即可：

• 修改Notebook的Block 2中ProcessingOutput的source：

  outputs=[
      ProcessingOutput(output_name="test_data", source="/opt/ml/processing/participant_relationship"),
  ],
  

• 或修改脚本中的输出路径为/opt/ml/processing/test。

2. 脚本未写入输出文件 #

脚本仅定义了输出路径变量，未执行df.to_csv()将数据写入文件，导致输出目录为空，SageMaker无法捕获输出。

修复：在脚本末尾添加写入逻辑：

# 创建输出目录（不存在则自动创建）
os.makedirs(os.path.dirname(parsed_suspect_relationship_output_path), exist_ok=True)
# 将处理后的数据写入CSV
df.to_csv(parsed_suspect_relationship_output_path, index=False)
print(f"输出文件已写入：{parsed_suspect_relationship_output_path}")

3. 缺少必要模块导入 #

脚本使用了re.findall()但未导入re模块，运行时会抛出NameError导致脚本崩溃。

修复：在脚本开头添加导入语句：

import re

4. 输入路径冗余定义 #

脚本先定义本地输入路径，又硬编码S3路径，忽略了Notebook配置的ProcessingInput映射，不符合SageMaker最佳实践。

修复：使用本地映射路径读取数据，移除硬编码的S3路径：

input_data_path = os.path.join("/opt/ml/processing/input", "gun-violence.csv") # 确保文件名与S3桶中一致
print("Reading input data from {}".format(input_data_path))
df = pd.read_csv(input_data_path)

总结 #

作业失败的核心原因是输出路径不匹配、脚本未写入输出文件、缺少模块导入，以及输入路径的不当处理。按上述步骤修复后，重新运行作业即可解决问题。

内容的提问来源于stack exchange，提问作者chongochoo