Lindorm Ray node

Step 1: Write the Python code

In the code editor for the Lindorm Ray node, enter the following Python code. The code uses the Ray @ray.remote decorator to distribute the computation across multiple workers for parallel execution, and estimates pi through random sampling.

import ray
import random
import time
import sys

ray.init()

@ray.remote
def compute_points_in_circle(num_points: int) -> int:
    """
    Remote function: Randomly generates points in a unit square and counts how many fall within the unit circle.
    """
    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:
    """
    Use Ray to estimate pi in parallel.
    """
    start_time = time.time()
    points_per_worker = total_points // num_workers

    print(f"Estimating pi with total points: {total_points:,}, workers: {num_workers}")

    # Submit multiple remote tasks.
    tasks = [compute_points_in_circle.remote(points_per_worker) for _ in range(num_workers)]

    # Collect the results.
    results = ray.get(tasks)
    total_inside = sum(results)

    # Calculate pi.
    pi_estimate = 4.0 * total_inside / total_points

    duration = time.time() - start_time

    print("=" * 60)
    print("Ray Sample: Calculating π using Monte Carlo Method")
    print("=" * 60)
    print(f"Total samples: {total_points:,}")
    print(f"Points inside circle: {total_inside:,}")
    print(f"Estimated π: {pi_estimate:.10f}")
    print(f"Time taken: {duration:.4f} seconds")
    print(f"Workers used: {num_workers}")
    print("=" * 60)
    print("Calculation complete!")

    return pi_estimate

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

Step 2: Configure the Ray submission command

In the Ray submission command area of the node, configure the following submission command.

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

Node content configuration

Step 1: Create and upload resource files

Create the following two Python resource files in DataWorks.

Resource file 1: circle_utils.py

"""
Utility module for calculating points within a circle.
"""
import random

def compute_points_in_circle(num_points: int) -> int:
    """
    Randomly generates points in a unit square and counts how many fall within the unit circle.
    """
    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

Resource file 2: pi_estimator.py

"""
Utility module for estimating pi.
"""
import time

def estimate_pi(results, total_points: int) -> float:
    """
    Estimates pi based on the number of points inside the circle.
    """
    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: {total_points:,}")
    print(f"Points inside circle: {total_inside:,}")
    print(f"Estimated pi ≈ {pi_estimate:.6f}")
    print(f"Time taken: {duration:.2f} seconds")

    return pi_estimate

Step 2: Write the main program code

In the code editor of the Lindorm Ray node, reference the uploaded resource files with ##@resource_reference and write the main program code.

##@resource_reference{"circle_utils.py"}
##@resource_reference{"pi_estimator.py"}
"""
Main program to calculate pi in parallel by using 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:
    """
    Remote function: A wrapper for the calculation function.
    """
    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"Estimating pi with total points: {total_points:,}, workers: {num_workers}")

    # Submit multiple remote tasks.
    tasks = [compute_points_in_circle_remote.remote(points_per_worker) for _ in range(num_workers)]

    # Get the results.
    results = ray.get(tasks)

    # Calculate pi.
    estimated_pi = estimate_pi(results, total_points)

    # Shut down Ray.
    ray.shutdown()

Step 3: Configure the Ray submission command

In the Ray submission command area of the node, configure the following submission command.

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