Streaming MapReduce — Ray v2.0.0.dev0 - What is Ray?

约定不等于承诺〃 2022-10-05 04:46 162阅读 0赞

> ⚠️ \[Ray Summit 2021 | June 22-24 | Virtual & Free\]
> 
> [Join the global Ray community at Ray Summit 2021][] to learn about new Ray features and hear how users are scaling machine learning applications with Ray!

# Streaming MapReduce¶ #

> This document walks through how to implement a simple streaming application using Ray’s actor capabilities. It implements a streaming MapReduce which computes word counts on wikipedia articles.

本文档介绍了如何使用 Ray 的 actor 功能实现一个简单的流应用程序。它实现了一个流式 MapReduce，它计算维基百科文章的字数。

### Overview ###

*  Streaming MapReduce¶
 *   *  Reference

--------------------

> You can view the [code for this example][].
> 
> To run the example, you need to install the dependencies

您可以查看此示例的代码。 要运行示例，您需要安装依赖项

pip install wikipedia

然后执行脚本如下：

python ray/doc/examples/streaming/streaming.py

对于每轮阅读的文章，脚本将输出这些文章中的前 10 个单词及其字数：

article index = 0
       the 2866
       of 1688
       and 1448
       in 1101
       to 593
       a 553
       is 509
       as 325
       are 284
       by 261
    article index = 1
       the 3597
       of 1971
       and 1735
       in 1429
       to 670
       a 623
       is 578
       as 401
       by 293
       for 285
    article index = 2
       the 3910
       of 2123
       and 1890
       in 1468
       to 658
       a 653
       is 488
       as 364
       by 362
       for 297
    article index = 3
       the 2962
       of 1667
       and 1472
       in 1220
       a 546
       to 538
       is 516
       as 307
       by 253
       for 243
    article index = 4
       the 3523
       of 1866
       and 1690
       in 1475
       to 645
       a 583
       is 572
       as 352
       by 318
       for 306
    ...

> Note that this examples uses [distributed actor handles][], which are still considered experimental.
> 
> There is a `Mapper` actor, which has a method `get_range` used to retrieve word counts for words in a certain range:

请注意，此示例使用分布式actor 句柄，仍被认为是实验性的。  
有一个 `Mapper` actor，它有一个方法 `get_range` 用于检索特定范围内单词的字数：

@ray.remote
    class Mapper(object):
    
        def __init__(self, title_stream):
            # Constructor, the title stream parameter is a stream of wikipedia
            # article titles that will be read by this mapper
    
        def get_range(self, article_index, keys):
            # Return counts of all the words with first
            # letter between keys[0] and keys[1] in the
            # articles that haven't been read yet with index
            # up to article_index

> The `Reducer` actor holds a list of mappers, calls `get_range` on them and accumulates the results.

`Reducer` actor 持有一个映射器列表，对它们调用 `get_range` 并累积结果。

@ray.remote
    class Reducer(object):
    
        def __init__(self, keys, *mappers):
             # Constructor for a reducer that gets input from the list of mappers
             # in the argument and accumulates word counts for words with first
             # letter between keys[0] and keys[1]
    
        def next_reduce_result(self, article_index):
             # Get articles up to article_index that haven't been read yet,
             # accumulate the word counts and return them

> On the driver, we then create a number of mappers and reducers and run the streaming MapReduce:

在驱动程序上，我们然后创建一些映射器和化简器并运行流式 MapReduce：

streams = # Create list of num_mappers streams
    keys = # Partition the keys among the reducers.
    
    # Create a number of mappers.
    mappers = [Mapper.remote(stream) for stream in streams]
    
    # Create a number of reduces, each responsible for a different range of keys.
    # This gives each Reducer actor a handle to each Mapper actor.
    reducers = [Reducer.remote(key, *mappers) for key in keys]
    
    article_index = 0
    while True:
        counts = ray.get([reducer.next_reduce_result.remote(article_index)
                          for reducer in reducers])
        article_index += 1

> The actual example reads a list of articles and creates a stream object which produces an infinite stream of articles from the list. This is a toy example meant to illustrate the idea. In practice we would produce a stream of non-repeating items for each mapper.

实际示例读取文章列表并创建一个流对象，该对象从列表中生成无限的文章流。这是一个玩具示例，旨在说明这个想法。在实践中，我们将为每个映射器生成一个非重复项的流。

--------------------

## Reference ##

https://docs.ray.io/en/master/auto\_examples/plot\_streaming.html

[Join the global Ray community at Ray Summit 2021]: https://www.anyscale.com/ray-summit-2021?utm_source=anyscale&utm_medium=docs&utm_campaign=raysummit
[code for this example]: https://github.com/ray-project/ray/tree/master/doc/examples/streaming
[distributed actor handles]: http://docs.ray.io/en/master/actors.html