实验 4：MapReduce 小项目

这个实验把前面的 Task 和 ObjectRef 组合起来，实现一个简单词频统计。

运行真实 Ray 示例

uv run python examples/ray_demos/04_map_reduce.py

核心代码：

@ray.remote
def count_words(chunk: str) -> Counter:
    words = [word.strip(".,;:!?()[]").lower() for word in chunk.split()]
    return Counter(word for word in words if word)

@ray.remote
def merge_counts(parts: list[Counter]) -> Counter:
    total = Counter()
    for part in parts:
        total.update(part)
    return total

partials = [count_words.remote(chunk) for chunk in chunks]
result = ray.get(merge_counts.remote(partials))

运行 Mini Ray 示例

uv run python examples/mini_ray_runtime/demos/04_map_reduce.py

Mini Ray 版本使用同样的 API 形状：

import mini_ray as ray

@ray.remote
def count_words(chunk):
    ...

@ray.remote
def merge_counts(parts):
    ...

这说明本教程实现的运行时已经能表达一个小型任务 DAG。

DAG 结构

为什么 merge_counts.remote(partials) 可行

partials 是一组 ObjectRef。Ray 看到它们作为参数传入，会把它们当作依赖。merge_counts 会在所有部分结果 ready 后运行。

这比你手写同步逻辑简单很多：

partial_values = ray.get(partials)
result = ray.get(merge_counts.remote(partial_values))

上面这种写法也能跑，但它把中间结果拉回 Driver，再传给 reduce，可能增加不必要的数据移动。

性能思考

词频统计是教学例子，不一定比本地单进程快。Ray 的收益取决于：

• 输入数据是否足够大。
• 每个 chunk 的计算是否足够重。
• 任务数量是否和资源匹配。
• 数据是否需要跨节点传输。
• reduce 是否成为瓶颈。

改造成树形 reduce

如果 partials 很多，单个 merge_counts 可能成为瓶颈。可以分层合并：

def tree_reduce(refs, fan_in=8):
    while len(refs) > 1:
        refs = [
            merge_counts.remote(refs[i : i + fan_in])
            for i in range(0, len(refs), fan_in)
        ]
    return refs[0]

树形 reduce 更适合大规模分片。

小练习

1. 把 chunk 大小改成 100、1000、10000，观察任务数量和耗时。
2. 实现树形 reduce。
3. 把输入文件换成一本较大的文本，比较本地 Counter 和 Ray 版本。
4. 在 Mini Ray 里打印 ray.metrics()，观察对象数量。