流水线读取MLTABLE时遇ScriptExecution.StreamAccess.NotFound错误求助

错误详情 #

Error Code: ScriptExecution.StreamAccess.NotFound
Native Error: error in streaming from input data sources
    StreamError(NotFound)
=> stream not found
    NotFound
Error Message: The requested stream was not found. Please make sure the request uri is correct.| session_id=1f8669ce-5a60-494b-a8dd-fd07fee8b186

背景说明 #

通过flight_data = ml_client.data.get(name='flightdelaydata1', version='2')获取数据后，交互式会话中可正常读取MLTABLE，代码如下：

import mltable
tbl = mltable.load(f'azureml:/{flight_data.id}')
tbl.to_pandas_dataframe()

但在流水线运行时，执行数据处理脚本却触发上述错误。以下是流水线中的相关代码：

数据处理脚本（data_prep.py） #

%%writefile {data_prep_folder}/data_prep.py
import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error
import mlflow
import mltable
import os

import argparse

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--data', help='Input data for flight delay model', type=str)
    parser.add_argument('--train_test_split_ratio', help='Test data proportion', type=float, default=.20)
    parser.add_argument('--train_data', help='Training Data', type=str)
    parser.add_argument('--test_data', help='Test Data', type=str)

    args = parser.parse_args()

    mlflow.start_run()

    tbl = mltable.load(args.data)
    df = tbl.to_pandas_dataframe()
    mlflow.log_metric('Number of observations', df.shape[0])
    mlflow.log_metric('Number of features', df.shape[1])

    df = df.dropna()
    df = df.loc[ : , ['Month', 'DayofMonth', 'DayOfWeek', 'DepDelay', 'DepDel15', 'ArrDel15', 'Cancelled', 'ArrDelay']]

    Train_data, Test_data = train_test_split(df, test_size=args.train_test_split_ratio)

    Train_data.to_csv(os.path.join(args.train_data, 'train.csv'))
    Test_data.to_csv(os.path.join(args.test_data, 'test.csv'))

    mlflow.end_run()

if __name__ == "__main__":
    main()

Command组件代码 #

from azure.ai.ml.constants import AssetTypes

data_prep_component = command(name='flight_delay_model_data_prep',
                              description='Flight Delay Model Prediction Data Preparation Component',
                              display_name='Flight Delay Data Prep',
                              inputs = {
                                        'data' : Input(type=AssetTypes.MLTABLE, path = flight_data.path),
                                        'train_test_split_ratio' : Input(type='number')
                              },
                              outputs = {
                                        'train_data' : Output(type = 'uri_folder'),
                                        'test_data' : Output(type = 'uri_folder')
                                        },
                              command= '''python {data_prep_folder}/data_prep.py \
                                            --data ${{inputs.data}} --train_test_split_ratio ${{inputs.train_test_split_ratio}} \
                                            --train_data ${{outputs.train_data}} --test_data ${{outputs.test_data}}''',
                              environment = f'{envt.name}:{envt.version}'
                              )

排查与解决方法 #

1. 修正MLTABLE资产引用格式 #

流水线中直接传递flight_data.path可能是原始存储路径（如ADLS路径），而非mltable.load()支持的MLTABLE资产引用格式。将Command组件的data输入路径修改为交互式会话中使用的格式：

'data' : Input(type=AssetTypes.MLTABLE, path = f'azureml:/{flight_data.id}'),

或直接传入资产ID：

'data' : Input(type=AssetTypes.MLTABLE, path = flight_data.id),

2. 检查计算环境权限 #

流水线使用的计算集群（服务主体/托管标识）可能缺少MLTABLE对应存储容器的访问权限，而交互式会话使用的是你的用户权限。需为计算集群的标识添加存储容器的Blob Data Contributor或对应RBAC角色。

3. 验证MLTABLE定义文件 #

重新注册MLTABLE资产，确保资产中的MLTable定义文件路径配置正确，无路径拼写或格式错误。

4. 打印输入路径排查 #

在data_prep.py的main函数中添加打印语句，输出实际接收的路径值，确认格式是否符合要求：

print(f"Input data path received: {args.data}")

查看流水线日志中的输出，判断路径是否为mltable.load()可识别的格式。

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