能否在Amazon SageMaker中对处理+训练流程做超参数优化？

可以实现，无需合并处理作业（Processing Job）与训练作业（Training Job）步骤，以下是具体实现方式：

核心思路 #

让每个超参数调优trial独立执行「处理→训练」流程，将处理作业作为训练作业的前置依赖，同时支持超参数传递给处理或训练步骤，确保Pipeline正确关联两者的输入输出逻辑。

具体实现步骤 #

1. 定义可接收超参数的处理作业步骤 #

编写处理脚本时，通过命令行参数读取可调优的预处理参数（如特征缩放方式、降维维度），示例核心逻辑：

import argparse
import pandas as pd
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.decomposition import PCA

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    # 固定输入输出路径
    parser.add_argument("--input-data", type=str, default="/opt/ml/processing/input")
    parser.add_argument("--output-data", type=str, default="/opt/ml/processing/output")
    # 可调优的预处理超参数
    parser.add_argument("--feature-scaling", type=str, default="standard")
    parser.add_argument("--n-components", type=int, default=10)
    
    args = parser.parse_args()
    
    # 读取原始数据
    data = pd.read_csv(f"{args.input_data}/raw_data.csv")
    X = data.drop("label", axis=1)
    y = data["label"]
    
    # 根据超参数执行预处理
    if args.feature_scaling == "standard":
        scaler = StandardScaler()
    else:
        scaler = MinMaxScaler()
    X_scaled = scaler.fit_transform(X)
    
    pca = PCA(n_components=args.n_components)
    X_processed = pca.fit_transform(X_scaled)
    
    # 保存处理后的数据
    processed_data = pd.DataFrame(X_processed)
    processed_data["label"] = y
    processed_data.to_csv(f"{args.output_data}/processed_data.csv", index=False)

在Pipeline中定义Processing Step时，将超参数设为可配置的参数对象：

from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep
from sagemaker.workflow.parameters import ParameterString, ParameterInteger

# 定义可调优的预处理参数
feature_scaling_param = ParameterString(name="FeatureScaling", default_value="standard")
n_components_param = ParameterInteger(name="NComponents", default_value=10)

processing_step = ProcessingStep(
    name="PreprocessingStep",
    processor=sklearn_processor,
    inputs=[ProcessingInput(source="s3://your-bucket/raw-data", destination="/opt/ml/processing/input")],
    outputs=[ProcessingOutput(source="/opt/ml/processing/output", destination="s3://your-bucket/processed-data")],
    job_arguments=[
        "--feature-scaling", feature_scaling_param,
        "--n-components", n_components_param
    ]
)

2. 定义依赖处理输出的训练作业步骤 #

训练作业的输入直接引用Processing Step的输出结果，同时定义训练相关的超参数占位符：

from sagemaker.workflow.steps import TrainingStep
from sagemaker.inputs import TrainingInput

# 训练超参数占位符
learning_rate_param = ParameterString(name="LearningRate", default_value="0.01")
batch_size_param = ParameterInteger(name="BatchSize", default_value=32)

training_step = TrainingStep(
    name="TrainingStep",
    estimator=xgb_estimator,
    inputs={
        "train": TrainingInput(
            s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["output"].S3Output.S3Uri,
            content_type="text/csv"
        )
    },
    hyperparameters={
        "learning_rate": learning_rate_param,
        "batch_size": batch_size_param,
        "n_estimators": 100
    }
)

3. 创建覆盖全流程的Tuner Step #

将预处理和训练的超参数都纳入搜索空间，让每个调优trial触发对应的处理和训练作业：

from sagemaker.tuner import HyperparameterTuner, CategoricalParameter, ContinuousParameter, IntegerParameter
from sagemaker.workflow.steps import TunerStep

# 定义超参数搜索范围
hyperparameter_ranges = {
    # 预处理超参数
    "FeatureScaling": CategoricalParameter(["standard", "minmax"]),
    "NComponents": IntegerParameter(5, 20),
    # 训练超参数
    "LearningRate": ContinuousParameter(0.001, 0.1),
    "BatchSize": CategoricalParameter([16, 32, 64])
}

# 初始化Tuner
tuner = HyperparameterTuner(
    estimator=xgb_estimator,
    objective_metric_name="validation:accuracy",
    hyperparameter_ranges=hyperparameter_ranges,
    max_jobs=10,
    max_parallel_jobs=3
)

# 创建Tuner Step，关联处理步骤和所有超参数
tuner_step = TunerStep(
    name="HyperparameterTuningStep",
    tuner=tuner,
    inputs={
        "train": TrainingInput(
            s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["output"].S3Output.S3Uri
        )
    },
    hyperparameters={
        "learning_rate": learning_rate_param,
        "batch_size": batch_size_param
    },
    parameters={
        "FeatureScaling": feature_scaling_param,
        "NComponents": n_components_param
    },
    depends_on=[processing_step]
)

4. 组装并执行Pipeline #

将所有步骤加入Pipeline并提交执行，每个调优trial会先运行对应参数的处理作业，再执行训练作业，实现完整的流程优化。

关键注意事项 #

• 若预处理参数无需调优，仅需固定执行一次处理作业，后续训练trial复用其输出，此时只需将Processing Step设为Tuner Step的上游依赖，无需传递预处理超参数。
• 为避免不同trial的处理结果互相覆盖，可在处理脚本中通过SageMaker环境变量SM_TRIAL_ID生成唯一输出路径。

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