如何在AzureML Python SDK v2中异步提交流水线不阻塞终端？

方案1：移除日志流方法 #

阻塞终端的核心原因是ml_client.jobs.stream(job.name)会持续输出作业日志直到任务完成，直接移除该行代码即可让终端立即释放。提交后可记录作业ID，后续按需查询状态或日志。

修改后的代码：

from azure.ai.ml.dsl import pipeline

@pipeline
def my_pipeline():
    ...

pipeline_obj = my_pipeline()
# 提交作业并返回作业对象
job = ml_client.jobs.create_or_update(pipeline_obj)
# 输出作业信息便于后续追踪
print(f"流水线已提交，作业ID: {job.name}，当前状态: {job.status}")

方案2：后台线程执行日志流（按需使用） #

如果需要实时查看日志但不想阻塞主进程，可通过Python多线程将日志流操作放到后台执行，主进程可继续处理其他任务。

示例代码：

import threading
from azure.ai.ml.dsl import pipeline

@pipeline
def my_pipeline():
    ...

pipeline_obj = my_pipeline()
job = ml_client.jobs.create_or_update(pipeline_obj)
print(f"流水线已提交，作业ID: {job.name}，当前状态: {job.status}")

# 定义日志流执行函数
def stream_job_logs(job_id):
    ml_client.jobs.stream(job_id)

# 启动后台线程运行日志流
log_thread = threading.Thread(target=stream_job_logs, args=(job.name,))
log_thread.start()

# 主进程可继续执行其他操作
print("主进程继续执行后续任务...")

方案3：定时查询作业状态 #

若无需实时日志，仅需追踪作业进度，可定时调用ml_client.jobs.get()查询状态，避免持续阻塞。

示例代码：

import time
from azure.ai.ml.dsl import pipeline

@pipeline
def my_pipeline():
    ...

pipeline_obj = my_pipeline()
job = ml_client.jobs.create_or_update(pipeline_obj)
job_id = job.name
print(f"流水线已提交，作业ID: {job_id}")

# 定时查询作业状态（可调整查询间隔）
while True:
    current_job = ml_client.jobs.get(job_id)
    print(f"当前作业状态: {current_job.status}")
    # 作业进入终态时停止查询
    if current_job.status in ["Completed", "Failed", "Canceled"]:
        print(f"作业结束，最终状态: {current_job.status}")
        break
    time.sleep(30)  # 每30秒查询一次

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