python並列処理ライブラリRayの使い方

使い方の結論

使い方は次のような感じ。説明はこのページの下のほうにあります。

import time
import random
from tqdm import tqdm
import ray

@ray.remote # 並列化するために関数をデコレート
def func(random_num, big_object):

    ######## 何かの計算 ########
    time.sleep(random_num)              # スリープ
    ############################

    return random_num                   # 何かを計算した体で、返却


def main():
    # funcの引数1
    random_num_list = [random.randint(0, 50) for _ in range(20)]

    # funcの引数2（とても大きいオブジェクトとする）
    big_object = None

    ######## 並列化処理 ########
    result_list = []
    ray.init()                     
    big_object_id = ray.put(big_object)         
    worker_ids = [func.remote(random_num, big_object_id) for random_num in random_num_list]    
    for _ in tqdm(range(len(random_num_list))):
        fin_worker_ids, worker_ids = ray.wait(worker_ids, num_returns=1)         
        result_list += ray.get(fin_worker_ids)                           
    ############################

    print(result_list)


if __name__ == '__main__':
    main()

rayのインストールについて

私はpip install -U rayでインストールしました。

2022/09/01現在、pipでインストールするのがよいようです。
https://docs.ray.io/en/latest/ray-overview/installation.html

使い方の解説

各行に対して、解説をコメントとして書いてみました。

import time
import random
from tqdm import tqdm
import ray

@ray.remote # 並列化するために関数をデコレート
def func(random_num, big_object):

    ######## 何かの計算 ########
    time.sleep(random_num)              # スリープ
    ############################

    return random_num                   # 何かを計算した体で、返却


def main():
    # funcの引数1
    random_num_list = [random.randint(0, 50) for _ in range(20)]

    # funcの引数2（とても大きいオブジェクトとする）
    big_object = None

    ######## 並列化処理 ########
    result_list = []

    # 初期化
    ray.init()                     

    # 各プロセスでの引数のコピーを防ぐために、先にrayにコピーさせて、そのコピーのidを取得
    big_object_id = ray.put(big_object)     
    
    # 並列処理を実行
    worker_ids = [func.remote(random_num, big_object_id) for random_num in random_num_list]
    
    # 計算が終わったプロセスから、計算結果を取得
    for _ in tqdm(range(len(random_num_list))):

        # num_returnsの数のプロセスが計算終了後、戻り値を取り出すために終わったプロセスを取得
        fin_worker_ids, worker_ids = ray.wait(worker_ids, num_returns=1) 
        
        # 計算が終わったプロセスが関数から受け取った戻り値を取得
        result_list += ray.get(fin_worker_ids)                           

    ############################

    print(result_list)


if __name__ == '__main__':
    main()

実行結果

上のコードを実行すると次のような出力が出るかと思います。

100%|█████████████████████████████████████████████████████████████████████████████████████████████| 20/20 [00:48<00:00,  2.40s/it]
[0, 2, 3, 4, 5, 6, 11, 12, 16, 18, 19, 24, 29, 31, 35, 40, 41, 43, 43, 48]

おわりに

Rayの解説記事少なくないですか…？