DatabricksにおけるRayサポートがGAになりました

約一年前にこちらの記事を書きました。

https://qiita.com/taka_yayoi/items/bd95bfe6457e0daa12bf

https://qiita.com/taka_yayoi/items/f919ad6aee56539f5309

**Rayとは**

https://docs.ray.io/en/latest/ray-overview/index.html

> RayはスケーラブルなAI、Pythonワークロードを実行するための有名な計算フレームワークであり、さまざまな分散機械学習ツール、大規模ハイパーパラメーターチューニング能力、強化学習アルゴリズム、モデルサービングなどを提供しています。

私は、Sparkはデータの並列分散が前提ですが、Rayを使うことでデータを伴わない処理を分散できると認識しています。

このたび、この機能がGAになりました！Databricksランタイム15.0以降ではデフォルトでRayがインストールされ、追加のインストールは不要になります。

https://www.databricks.com/blog/announcing-general-availability-ray-databricks

マニュアルはこちら。

https://docs.databricks.com/ja/machine-learning/ray-integration.html

サンプルノートブックをウォークスルーします。

クラスターはノートブックの要件に沿って、4GPU * 2ノードのGPUクラスターを準備します。ランタイムは15.0 MLを指定します。
![Screenshot 2024-04-22 at 9.47.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6314dea7-76aa-e123-20ae-debd220c8f6e.png)

# Rayのインストール

ランタイム15.0 MLの場合、Rayの追加インストールは不要です。
![Screenshot 2024-04-22 at 9.47.35.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6e28239e-ab86-36c0-b349-c71505cc1b35.png)

# Rayクラスターの作成

2台のRayワーカーノードを持つRayクラスターをセットアップします。それぞれのワーカーノードには4個のGPUコアが割り当てられています。クラスターが起動すると、Rayクラスターダッシュボードを参照するために、リンク"Open Ray Cluster Dashboard in a new tab"をクリックすることができます。

```py
from ray.util.spark import setup_ray_cluster, shutdown_ray_cluster, MAX_NUM_WORKER_NODES

setup_ray_cluster(
  num_worker_nodes=2,
  num_cpus_per_node=4,
  collect_log_to_path="/dbfs/tmp/raylogs",
)
```
![Screenshot 2024-04-22 at 9.49.09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/75e148f4-6abc-e1b7-f420-6cc089694bee.png)

作成したRayクラスターのリソースを表示します。

```
{'object_store_memory': 24466633522.0,
 'CPU': 8.0,
 'accelerator_type:T4': 2.0,
 'memory': 55657155790.0,
 'GPU': 2.0,
 'node:10.0.7.67': 1.0,
 'node:__internal_head__': 1.0,
 'node:10.0.6.78': 1.0,
 'node:10.0.6.91': 1.0}
```

# Rayアプリケーションの実行

以下のコードでは、先程作成したRayクラスターで多数のRayタスクを作成するシンプルなRayアプリケーションを実行します。

```py
import ray
import random
import time
import math
from fractions import Fraction

@ray.remote
def pi4_sample(sample_count):
    """pi4_sampleはsample_count回の実験を実行し、
    円の中に含まれた回数の比率を返却します。
    """
    in_count = 0
    for i in range(sample_count):
        x = random.random()
        y = random.random()
        if x*x + y*y <= 1:
            in_count += 1
    return Fraction(in_count, sample_count)

SAMPLE_COUNT = 1000 * 1000
start = time.time() 
future = pi4_sample.remote(sample_count = SAMPLE_COUNT)
pi4 = ray.get(future)
end = time.time()
dur = end - start
print(f'Running {SAMPLE_COUNT} tests took {dur} seconds')

pi = pi4 * 4
print(float(pi))
```

```
Running 1000000 tests took 0.8532092571258545 seconds
3.14098
```

上で表示されたリンクをクリックすることで、Rayダッシュボードにアクセスできます。
![Screenshot 2024-04-22 at 9.49.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e72cf497-4054-f840-a838-bd95eed750fa.png)
![Screenshot 2024-04-22 at 9.52.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/e549cdd9-ce36-7070-26b1-8c78ec536f30.png)

```py
FULL_SAMPLE_COUNT = 2000 * 1000 * 1000
BATCHES = int(FULL_SAMPLE_COUNT / SAMPLE_COUNT)
print(f'Doing {BATCHES} batches')
results = []
for _ in range(BATCHES):
    results.append(pi4_sample.remote(sample_count = SAMPLE_COUNT))
output = ray.get(results)

pi = sum(output)*4/len(output)
print(float(pi))
```

```
Doing 2000 batches
3.141600784
```

# Rayクラスターのシャットダウン

```py
shutdown_ray_cluster()
```

ユースケース見ると、LLMとかでも使えるようですね。トライしてみます。

https://docs.ray.io/en/latest/ray-overview/use-cases.html

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)