背景

Ray Dashboard 提供了基础集群监控功能，但在复杂分布式 AI 计算场景中仍有局限：

• 开发者很难全面了解系统的执行流程和数据流向，进而很难识别分布式环境中的性能瓶颈。
• 无法获得统一的资源利用率视图，进而难以提高资源利用率。
• 日志和调试信息通常分散在多个节点上难以收集和关联，进而调试和分析问题异常困难。

Flow Insight 视图为 Ray 应用的执行流程提供了强大的可视化功能，帮助用户理解分布式应用中 actors、tasks 和数据传输之间的关系。该视图提供了应用执行的逻辑视图、物理视图、分布式栈、火焰图、甘特图等五种可视化模式，便于分析性能、识别瓶颈并优化 Ray 工作负载。

• 逻辑视图(Logical View)：展示组件间的调用关系和数据依赖，帮助理解系统拓扑结构。
• 物理视图(Physical View)：展示组件在节点间的分布和资源使用情况，帮助优化部署策略。
• 分布式调用栈(Distributed Stack)：提供跨节点的调用链路追踪，快速定位执行卡点。
• 分布式火焰图(Distributed Flame Graph)：可视化系统性能热点分布，精准定位性能瓶颈。
• 甘特图(Gantt Chart)：展示函数和 actor 的执行时间序列。

本文将通过一个具体示例，指导您如何为 Serverless Ray 作业开启并使用 Flow Insight 功能进行可视化分析。

Flow Insight 示例

代码准备

请将以下代码保存为 test_flowinsight.py 文件：

• 代码示例。

import ray
import random
import time
import os

@ray.remote
class Worker:
    """
    Create a tree of workers in which each worker spawns multiple children.
    """
    def __init__(self, depth, is_bottleneck=False):
        # Whether this worker should mock delay
        self.is_bottleneck = is_bottleneck
        # Current depth in the worker tree (root = highest number)
        self.depth = depth
        # List to store child worker actor references
        self.children = []
        
        # Create child workers if we're not at a leaf node (depth > 0)
        if self.depth > 0:
            # Create 1-2 children randomly for each worker
            for _ in range(random.randint(1, 2)):
                # Recursively create child workers with decreased depth
                # Randomly decide if the child will be a bottleneck
                self.children.append(
                    Worker.remote(
                        depth=self.depth - 1,
                        is_bottleneck=random.randint(0, 1) == 1
                    )
                )
    def process(self):
        """
        Process work and delegate to child workers.
        Bottleneck workers at leaf nodes introduce significant delays.
        """
        # List to collect all async task references from children
        futures = []
        
        # For each child worker
        for child in self.children:
            # Submit 10 process tasks to each child
            for _ in range(10):
                # If this is a bottleneck worker at a leaf node (or just above),
# introduce a significant delay (30 seconds)
                if self.is_bottleneck and self.depth - 1 <= 0:
                    time.sleep(30.0)
                
# Standard processing time for all workers (5 seconds)
                time.sleep(5)
                
                # Submit the process task to the child worker
                # This creates a nested tree of task calls
                futures.append(child.process.remote())
        # Wait for all child tasks to complete before returning
        # This ensures the full tree of tasks must complete
        ray.get(futures)
# Create the root worker with depth 3
# This will create a tree of workers with maximum depth of 3
worker = Worker.remote(depth=3, is_bottleneck=False)
# Start processing and wait for the entire tree to complete
ray.get(worker.process.remote())

• 将 test_flowinsight.py 文件上传到 TOS。

# 代码打包
zip test_flowinsight.zip test_flowinsight.py
# 上传到tos路径，示例：
tos://tos-snzhao/serverless/test_flowinsight.zip

可视化作业提交

1. 进入 EMR Serverless 控制台＞队列名称＞队列详情＞作业提交，选择 RayJob 开发类型。
2. 在编辑作业页面，填写作业信息，点击运行。