Kubeflow Pipeline训练组件报错Unknown return type排查解决

触发原因 #

该错误是Kubeflow Pipelines组件编译器解析组件函数时抛出的，核心原因是组件函数存在未声明类型的返回值，编译器无法识别合法输出类型。你的代码中直接触发该错误的是内部函数缩进错误：

• 定义的onehot_encode工具函数中，return df语句没有缩进，不属于onehot_encode的内部逻辑，反而被识别为外层组件函数build_xgb_xgboost的返回语句。由于你没有为组件函数声明该返回值的类型，编译器检测到返回类型为inspect._empty（即无类型标注），直接抛出对应错误。

除此之外你的代码还存在4个会导致后续运行失败的隐藏问题：

• 计算RMSE时调用了np.sqrt，但未导入numpy库
• 保存模型时调用了dump方法，但未从joblib中导入该方法
• XGBRegressor的learning_rate、alpha参数传入了字符串格式的数值，会引发训练参数异常
• 入参bq_location声明后未使用，不符合组件定义规范
• 依赖列表中写的sklearn为无效PyPI包名，实际包名为scikit-learn，会导致依赖安装失败
• XGBoost旧版参数objective='reg:linear'已在新版本废弃，会触发训练告警

修复步骤 #

1. 修正onehot_encode函数的缩进，将return df移入函数内部，和其他逻辑对齐，避免组件函数意外返回未标注类型的对象
2. 补充缺失的导入语句：在组件内的导入块添加import numpy as np、from joblib import dump
3. 修正XGBRegressor的参数类型，将字符串格式的数值改为浮点数，同时替换废弃的objective参数
4. 修正依赖列表中的包名，将sklearn改为scikit-learn，补充numpy依赖
5. （可选）移除未使用的bq_location入参，或在初始化BigQuery客户端时传入该参数

修复后的完整核心代码 #

@component(
    packages_to_install=["google-cloud-bigquery", "xgboost", "pandas", "scikit-learn", "joblib", "pyarrow", "db_dtypes", "numpy"],
    base_image="python:3.9",
    output_component_file="create_xgb_model_xgboost.yaml"
)
def build_xgb_xgboost(project: str, 
                            bq_dataset: str, 
                            test_view_name: str,
                            metrics: Output[Metrics],
                            model: Output[Model]

):
    from google.cloud import bigquery
    import xgboost as xgb
    import pandas as pd
    import numpy as np
    from xgboost import XGBRegressor
    from sklearn.model_selection import train_test_split
    from sklearn.preprocessing import StandardScaler
    from sklearn.metrics import mean_squared_error as MSE
    from sklearn.metrics import mean_absolute_error
    from joblib import dump
    import pyarrow
    import db_dtypes
     

    client = bigquery.Client(project=project) 

    view_uri = f"{project}.{bq_dataset}.{test_view_name}"
    
    build_df_for_xgboost = '''
    SELECT * FROM `{view_uri}`
    '''.format(view_uri = view_uri)

    job_config = bigquery.QueryJobConfig()
    df_1 = client.query(build_df_for_xgboost).to_dataframe()
    
    df = df_1.drop(['int64_field_0'], axis=1)
    
    def onehot_encode(df, column):
        df = df.copy()
        dummies = pd.get_dummies(df[column], prefix=column)
        df = pd.concat([df, dummies], axis=1)
        df = df.drop(column, axis=1)
        return df # 修正缩进，移入函数内部
    
    # Binary encoding
    df['preferred_foot'] = df['preferred_foot'].replace({'left': 0, 'right': 1})
    
    # One-hot encoding
    for column in ['attacking_work_rate', 'defensive_work_rate']:
        df = onehot_encode(df, column=column)
    
    # Split df into X and y
    y = df['overall_rating']
    X = df.drop('overall_rating', axis=1)
    
    # Train-test split
    X_train, X_test, y_train, y_test = train_test_split(X, y, train_size=0.7, shuffle=True, random_state=1)
    
    # Scale X
    scaler = StandardScaler()
    scaler.fit(X_train)
    X_train = pd.DataFrame(scaler.transform(X_train), index=X_train.index, columns=X_train.columns)
    X_test = pd.DataFrame(scaler.transform(X_test), index=X_test.index, columns=X_test.columns)
    
    #define your model 
    bst = XGBRegressor(
    objective='reg:squarederror', # 替换已废弃的reg:linear参数
    learning_rate = 0.1, # 修正为数值类型
    alpha = 0.001 # 修正为数值类型
    )
    
    #fit your model
    bst.fit(X_train, y_train)
    
    # Predict the model 
    y_pred = bst.predict(X_test)
    rmse = np.sqrt(np.mean((y_test - y_pred)**2))
    mae = mean_absolute_error(y_test, y_pred)
    
    metrics.log_metric("RMSE", rmse)
    metrics.log_metric("framework", "xgboost")
    metrics.log_metric("dataset_size", len(df))
    metrics.log_metric("MAE", mae)
    
    dump(bst, model.path + ".joblib")

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