Kubeflow Pipeline组件步骤中混淆矩阵可视化不显示问题排查

我参照Kubeflow官方教程编写了如下代码，尝试在Kubeflow Pipeline组件步骤中渲染混淆矩阵可视化，但运行后，在Run输出和Visualizations标签页均无法看到可视化内容，页面提示“该步骤没有可视化”，请问我遗漏了什么？

import kfp
import kfp.dsl as dsl
from kfp.components import create_component_from_func    

@create_component_from_func
def confusion_visualization(matrix_uri: str = 'https://raw.githubusercontent.com/kubeflow/pipelines/master/samples/core/visualization/confusion_matrix.csv') -> NamedTuple('VisualizationOutput', [('mlpipeline_ui_metadata', 'UI_metadata')]):
    """Provide confusion matrix csv file to visualize as metrics."""
    import json

    metadata = {
        'outputs' : [{
          'type': 'confusion_matrix',
          'format': 'csv',
          'schema': [
            {'name': 'target', 'type': 'CATEGORY'},
            {'name': 'predicted', 'type': 'CATEGORY'},
            {'name': 'count', 'type': 'NUMBER'},
          ],
          'source': matrix_uri,
          'labels': ['rose', 'lily', 'iris'],
        }]
    }
    
    print('Printing the metadata')
    print(metadata)

    from collections import namedtuple
    visualization_output = namedtuple('VisualizationOutput', [
        'mlpipeline_ui_metadata'])
    print()
    return visualization_output(json.dumps(metadata))

@dsl.pipeline(
    name='confusion-matrix-pipeline',
    description='A sample pipeline to generate Confusion Matrix for UI visualization.'
)
def confusion_matrix_pipeline():
    confusion_visualization_task = confusion_visualization('results.json')
    
    
client = kfp.Client()
client.create_run_from_pipeline_func(
    confusion_matrix_pipeline,
    arguments={}
)

问题分析与解决方法 #

1. 数据源不符合要求 #

你在流水线中传入的results.json存在两个问题：

• 混淆矩阵组件要求数据源为CSV格式，而非JSON；
• 该文件在组件运行的容器环境中不存在，组件无法读取有效数据。

解决：使用官方示例的公开CSV链接，或者确保你的CSV文件可被组件访问（比如通过PVC挂载、存储卷映射，或者使用可公开访问的URL）。

2. NamedTuple类型注解缺失 #

函数签名中使用的NamedTuple未提前导入，虽然函数内部导入了collections.namedtuple，但类型注解需要typing.NamedTuple的支持，否则会导致输出元数据无法被Kubeflow正确识别。

解决：在代码顶部添加导入语句：

from typing import NamedTuple

3. 输出元数据匹配验证 #

确保labels列表的长度与CSV文件中的类别数量完全匹配，同时验证mlpipeline_ui_metadata的JSON格式无语法错误（可通过打印内容检查）。

修正后的完整代码示例 #

import kfp
import kfp.dsl as dsl
from kfp.components import create_component_from_func
from typing import NamedTuple  # 新增类型注解导入

@create_component_from_func
def confusion_visualization(matrix_uri: str = 'https://raw.githubusercontent.com/kubeflow/pipelines/master/samples/core/visualization/confusion_matrix.csv') -> NamedTuple('VisualizationOutput', [('mlpipeline_ui_metadata', 'UI_metadata')]):
    """Provide confusion matrix csv file to visualize as metrics."""
    import json

    metadata = {
        'outputs' : [{
          'type': 'confusion_matrix',
          'format': 'csv',
          'schema': [
            {'name': 'target', 'type': 'CATEGORY'},
            {'name': 'predicted', 'type': 'CATEGORY'},
            {'name': 'count', 'type': 'NUMBER'},
          ],
          'source': matrix_uri,
          'labels': ['rose', 'lily', 'iris'],
        }]
    }
    
    print('Printing the metadata')
    print(metadata)

    from collections import namedtuple
    visualization_output = namedtuple('VisualizationOutput', [
        'mlpipeline_ui_metadata'])
    print()
    return visualization_output(json.dumps(metadata))

@dsl.pipeline(
    name='confusion-matrix-pipeline',
    description='A sample pipeline to generate Confusion Matrix for UI visualization.'
)
def confusion_matrix_pipeline():
    # 使用官方示例的有效CSV链接，可替换为自己的合法数据源
    confusion_visualization_task = confusion_visualization()
    
    
client = kfp.Client()
client.create_run_from_pipeline_func(
    confusion_matrix_pipeline,
    arguments={}
)

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