DataWorks:Lindorm Ray ノード

ステップ 1：Python コードの記述

Lindorm Ray ノードのコードエディターに、次の Python コードを記述します。このコードは @ray.remote デコレーターを使用して、サンプリング計算を複数のワーカーに分散させます。

import ray
import random
import time
import sys

ray.init()

@ray.remote
def compute_points_in_circle(num_points: int) -> int:
    """
    単位正方形内にランダムな点を生成し、単位円内に収まる点の数をカウントするリモート関数。
    """
    inside = 0
    for _ in range(num_points):
        x = random.uniform(0, 1)
        y = random.uniform(0, 1)
        if x * x + y * y <= 1:
            inside += 1
    return inside

def estimate_pi(total_points: int, num_workers: int = 4) -> float:
    """
    Ray を使用して円周率 (π) を並列で推定します。
    """
    start_time = time.time()
    points_per_worker = total_points // num_workers

    print(f"円周率 (π) の推定を開始します。総ポイント数：{total_points:,}、ワーカー数：{num_workers}")

    # 複数のリモートタスクを送信します。
    tasks = [compute_points_in_circle.remote(points_per_worker) for _ in range(num_workers)]

    # 結果を収集します。
    results = ray.get(tasks)
    total_inside = sum(results)

    # 円周率 (π) を計算します。
    pi_estimate = 4.0 * total_inside / total_points

    duration = time.time() - start_time

    print("=" * 60)
    print("Ray サンプル：モンテカルロ法による円周率 (π) の計算")
    print("=" * 60)
    print(f"総サンプル数：{total_points:,}")
    print(f"円内のポイント数：{total_inside:,}")
    print(f"推定された円周率 (π)：{pi_estimate:.10f}")
    print(f"所要時間：{duration:.4f} 秒")
    print(f"使用されたワーカー数：{num_workers}")
    print("=" * 60)
    print("計算完了！")

    return pi_estimate

if __name__ == "__main__":
    estimated_pi = estimate_pi(total_points=int(sys.argv[1]), num_workers=4)
    ray.shutdown()

ステップ 2：送信コマンドの設定

ノードの [Ray 送信コマンド] セクションで、次の送信コマンドを設定します。

ray job submit \ 
--working-dir "." \
-- python calculate_pi.py 1000000

ノード設定

次のパラメーターの説明に基づいて、Lindorm Ray ノードを設定します。

ステップ 1：リソースファイルの作成とアップロード

DataWorks で、次の 2 つの Python リソースファイルを作成します。

リソースファイル 1：circle_utils.py

"""
円内のポイントを計算するためのユーティリティモジュール。
"""
import random

def compute_points_in_circle(num_points: int) -> int:
    """
    単位正方形内にランダムな点を生成し、単位円内に収まる点の数をカウントします。
    """
    inside = 0
    for _ in range(num_points):
        x = random.uniform(0, 1)
        y = random.uniform(0, 1)
        if x * x + y * y <= 1:
            inside += 1
    return inside

リソースファイル 2：pi_estimator.py

"""
円周率 (π) を推定するためのユーティリティモジュール。
"""
import time

def estimate_pi(results, total_points: int) -> float:
    """
    円内のポイント数に基づいて円周率 (π) を推定します。
    """
    start_time = time.time()
    total_inside = sum(results)
    pi_estimate = 4.0 * total_inside / total_points
    duration = time.time() - start_time

    print(f"総ポイント数：{total_points:,}")
    print(f"円内のポイント数：{total_inside:,}")
    print(f"推定された円周率 (π) ≈ {pi_estimate:.6f}")
    print(f"所要時間：{duration:.2f} 秒")

    return pi_estimate

ステップ 2：メインプログラムコードの記述

Lindorm Ray ノードのコードエディターで、##@resource_reference を使用してアップロードされたリソースファイルを参照し、メインプログラムのコードを記述します。

##@resource_reference{"circle_utils.py"}
##@resource_reference{"pi_estimator.py"}
"""
Ray を使用して円周率 (π) を並列で計算するメインプログラム。
"""
import sys
import ray
from circle_utils import compute_points_in_circle
from pi_estimator import estimate_pi

@ray.remote
def compute_points_in_circle_remote(num_points: int) -> int:
    """
    計算関数をラップするリモート関数。
    """
    return compute_points_in_circle(num_points)

if __name__ == "__main__":
    ray.init()

    total_points = int(sys.argv[1])
    num_workers = 4
    points_per_worker = total_points // num_workers

    print(f"円周率 (π) の推定を開始します。総ポイント数：{total_points:,}、ワーカー数：{num_workers}")

    # 複数のリモートタスクを送信します。
    tasks = [compute_points_in_circle_remote.remote(points_per_worker) for _ in range(num_workers)]

    # 結果を取得します。
    results = ray.get(tasks)

    # 円周率 (π) を計算します。
    estimated_pi = estimate_pi(results, total_points)

    # Ray をシャットダウンします。
    ray.shutdown()

ステップ 3：送信コマンドの設定

ノードの [Ray 送信コマンド] セクションで、次の送信コマンドを設定します。

ray job submit \
--working-dir "." \
-- python main.py 1000000