Ray: Verteiltes Computing für alle, Teil 2

Teil meiner zweiteiligen Serie über die Ray-Bibliothek, ein von AnyScale erstelltes Python-Framework für verteiltes und paralleles Rechnen. Teil 1 behandelt, wie Sie CPU-intensive Python-Jobs auf Ihrem lokalen PC parallelisieren, indem Sie die Arbeitslast auf alle verfügbaren Kerne verteilen, was zu deutlichen Laufzeitverbesserungen führt. Ich werde am Ende dieses Artikels einen Hyperlink zu Teil 1 hinterlassen.

Dieser Teil befasst sich mit einem ähnlichen Thema, außer dass wir die Verteilung von Python-Workloads auf die nächste Ebene bringen, indem wir Ray verwenden, um sie über Multi-Server-Cluster in der Cloud zu parallelisieren.

Wenn Sie dazu gekommen sind, ohne Teil 1 gelesen zu haben, ist die TL;DR von Ray, dass es sich um eine handelt Open-Supply verteiltes Pc-Framework entworfen, um es zu schaffen leicht zu skalieren Python-Programme von einem Laptop computer zu einem Cluster mit minimalen Codeänderungen. Das allein sollte hoffentlich ausreichen, um Ihr Interesse zu wecken. In meinem eigenen Take a look at auf meinem Desktop-PC habe ich ein unkompliziertes, relativ einfaches Python-Programm zum Finden von Primzahlen verwendet und seine Laufzeit um den Faktor 10 reduziert, indem ich nur vier Codezeilen hinzugefügt habe.

Wo können Sie Ray-Cluster betreiben?

Ray-Cluster können wie folgt eingerichtet werden:

• AWS und GCP Cloud, obwohl es auch inoffizielle Integrationen für andere Anbieter gibt, wie zum Beispiel Azure
• AnyScale, eine vollständig verwaltete Plattform, die von den Entwicklern von Ray entwickelt wurde.
• Kubernetes kann auch über das offiziell unterstützte KubeRay-Projekt genutzt werden.

Voraussetzungen

Um meinem Prozess folgen zu können, müssen Sie im Voraus einige Dinge einrichten. Ich werde AWS für meine Demo verwenden, da ich dort bereits über ein Konto verfüge; Ich gehe jedoch davon aus, dass die Einrichtung für andere Cloud-Anbieter und Plattformen sehr ähnlich sein wird. Sie sollten Folgendes haben:

• Anmeldeinformationen, die zum Ausführen von Cloud-CLI-Befehlen des von Ihnen gewählten Anbieters eingerichtet sind.
• Eine Customary-VPC und mindestens ein damit verbundenes öffentliches Subnetz, das über eine öffentlich erreichbare IP-Adresse verfügt.
• Eine SSH-Schlüsselpaardatei (.pem), die Sie auf Ihr lokales System herunterladen können, damit Ray (und Sie) eine Verbindung zu den Knoten in Ihrem Cluster herstellen können
• Sie verfügen über genügend Kontingente, um die angeforderte Anzahl von Knoten und vCPUs in dem von Ihnen eingerichteten Cluster zu erfüllen.

Wenn Sie Ihren Ray-Code lokal testen möchten, bevor Sie ihn in einem Cluster bereitstellen, müssen Sie auch die Ray-Bibliothek installieren. Wir können das mit pip machen.

$ pip set up ray

Ich werde alles über eine WSL2-Ubuntu-Shell auf meinem Home windows-Desktop ausführen.

Um zu überprüfen, ob Ray korrekt installiert wurde, sollten Sie den Befehlszeileninterpreter verwenden können. Geben Sie in einem Terminalfenster den folgenden Befehl ein.

$ ray --help

Utilization: ray (OPTIONS) COMMAND (ARGS)...

Choices:
  --logging-level TEXT   The logging stage threshold, decisions=('debug',
                         'data', 'warning', 'error', 'essential'),
                         default='data'
  --logging-format TEXT  The logging format.
                         default="%%(asctime)st%%(levelname)s
                         %%(filename)s:%%(lineno)s -- %%(message)s"
  --version              Present the model and exit.
  --help                 Present this message and exit.

Instructions:
  connect               Create or connect to a SSH session to a Ray cluster.
  check-open-ports     Examine open ports within the native Ray cluster.
  cluster-dump         Get log information from a number of nodes.
...
...
...

Wenn Sie dies nicht sehen, ist ein Fehler aufgetreten und Sie sollten die Ausgabe Ihres Installationsbefehls noch einmal überprüfen.

Vorausgesetzt, alles ist in Ordnung, kann es losgehen.

Ein letzter wichtiger Punkt jedoch. Erstellen von Ressourcen wie Rechenclustern bei einem Cloud-Anbieter wie AWS werden Kosten entstehenDaher ist es wichtig, dass Sie dies im Hinterkopf behalten. Die gute Nachricht ist, dass Ray über einen integrierten Befehl verfügt, der jede von Ihnen erstellte Infrastruktur zerstört. Um jedoch sicher zu gehen, sollten Sie noch einmal sicherstellen, dass keine ungenutzten und möglicherweise kostspieligen Dienste übrig bleiben „eingeschaltet“ versehentlich.

Unser Beispiel-Python-Code

Der erste Schritt besteht darin, unseren vorhandenen Ray-Code aus Teil 1 so zu ändern, dass er auf einem Cluster läuft. Hier ist der Originalcode als Referenz. Denken Sie daran, dass wir versuchen, die Anzahl der Primzahlen innerhalb eines bestimmten Zahlenbereichs zu zählen.

import math
import time

# -----------------------------------------
# Change No. 1
# -----------------------------------------
import ray
ray.init()

def is_prime(n: int) -> bool:
    if n < 2: return False
    if n == 2: return True
    if n % 2 == 0: return False
    r = int(math.isqrt(n)) + 1
    for i in vary(3, r, 2):
        if n % i == 0:
            return False
    return True

# -----------------------------------------
# Change No. 2
# -----------------------------------------
@ray.distant(num_cpus=1)  # pure-Python loop → 1 CPU per job
def count_primes(a: int, b: int) -> int:
    c = 0
    for n in vary(a, b):
        if is_prime(n):
            c += 1
    return c

if __name__ == "__main__":
    A, B = 10_000_000, 20_000_000
    total_cpus = int(ray.cluster_resources().get("CPU", 1))

    # Begin "chunky"; we are able to sweep this later
    chunks = max(4, total_cpus * 2)
    step = (B - A) // chunks

    print(f"nodes={len(ray.nodes())}, CPUs~{total_cpus}, chunks={chunks}")
    t0 = time.time()
    refs = ()
    for i in vary(chunks):
        s = A + i * step
        e = s + step if i < chunks - 1 else B
        # -----------------------------------------
        # Change No. 3
        # -----------------------------------------
        refs.append(count_primes.distant(s, e))

    # -----------------------------------------
    # Change No. 4
    # -----------------------------------------
    complete = sum(ray.get(refs))

    print(f"complete={complete}, time={time.time() - t0:.2f}s")

Welche Änderungen sind erforderlich, um es auf einem Cluster auszuführen? Die Antwort lautet: nur eine kleine Änderung ist erforderlich.

Change 

ray.init() 

to

ray.init(tackle=auto)

Das ist eine der Schönheiten von Ray. Derselbe Code läuft nahezu unverändert auf Ihrem lokalen PC und überall dort, wo Sie ihn ausführen möchten, einschließlich großer Cloud-Cluster mit mehreren Servern.

Aufbau unseres Clusters

In der Cloud besteht ein Ray-Cluster aus einem Hauptknoten und einem oder mehreren Employee-Knoten. In AWS sind alle diese Knoten einfach EC2-Instanzen. Ray-Cluster können eine feste Größe haben oder basierend auf den Ressourcen, die von den auf dem Cluster ausgeführten Anwendungen angefordert werden, automatisch nach oben und unten skaliert werden. Der Hauptknoten wird zuerst gestartet und die Employee-Knoten werden mit der Adresse des Hauptknotens konfiguriert, um den Cluster zu bilden. Wenn die automatische Skalierung aktiviert ist, skalieren Employee-Knoten basierend auf der Auslastung der Anwendung automatisch nach oben oder unten und werden nach einem vom Benutzer festgelegten Zeitraum (standardmäßig 5 Minuten) herunterskaliert.

Ray verwendet YAML-Dateien, um Cluster einzurichten. Eine YAML-Datei ist lediglich eine reine Textdatei mit einer JSON-ähnlichen Syntax, die für die Systemkonfiguration verwendet wird.

Hier ist die YAML-Datei, die ich zum Einrichten meines Clusters verwenden werde. Ich habe festgestellt, dass die EC2-Instanz, die meinem Desktop-PC in Bezug auf die Anzahl und Leistung der CPU-Kerne am nächsten kommt, eine c7g.8xlarge warfare. Der Einfachheit halber verwende ich den Hauptknoten als denselben Servertyp wie alle Employee, Sie können jedoch bei Bedarf verschiedene EC2-Typen kombinieren.

cluster_name: ray_test

supplier:
  sort: aws
  area: eu-west-1
  availability_zone: eu-west-1a

auth:
  # For Amazon Linux AMIs the SSH person is 'ec2-user'.
  # In case you change to an Ubuntu AMI, change this to 'ubuntu'.
  ssh_user: ec2-user
  ssh_private_key: ~/.ssh/ray-autoscaler_eu-west-1.pem

max_workers: 10
idle_timeout_minutes: 10

head_node_type: head_node

available_node_types:
  head_node:
    node_config:
      InstanceType: c7g.8xlarge
      ImageId: ami-06687e45b21b1fca9
      KeyName: ray-autoscaler_eu-west-1

  worker_node:
    min_workers: 5
    max_workers: 5
    node_config:
      InstanceType: c7g.8xlarge
      ImageId: ami-06687e45b21b1fca9
      KeyName: ray-autoscaler_eu-west-1
      InstanceMarketOptions:
        MarketType: spot

# =========================
# Setup instructions (run on head + employees)
# =========================
setup_commands:
  - |
    set -euo pipefail

    have_cmd() { command -v "$1" >/dev/null 2>&1; }
    have_pip_py() {
      python3 -c 'import importlib.util, sys; sys.exit(0 if importlib.util.find_spec("pip") else 1)'
    }

    # 1) Guarantee Python 3 is current
    if ! have_cmd python3; then
      if have_cmd dnf; then
        sudo dnf set up -y python3
      elif have_cmd yum; then
        sudo yum set up -y python3
      elif have_cmd apt-get; then
        sudo apt-get replace -y
        sudo apt-get set up -y python3
      else
        echo "No supported bundle supervisor discovered to put in python3." >&2
        exit 1
      fi
    fi

    # 2) Guarantee pip exists
    if ! have_pip_py; then
      python3 -m ensurepip --upgrade >/dev/null 2>&1 || true
    fi
    if ! have_pip_py; then
      if have_cmd dnf; then
        sudo dnf set up -y python3-pip || true
      elif have_cmd yum; then
        sudo yum set up -y python3-pip || true
      elif have_cmd apt-get; then
        sudo apt-get replace -y || true
        sudo apt-get set up -y python3-pip || true
      fi
    fi
    if ! have_pip_py; then
      curl -fsS https://bootstrap.pypa.io/get-pip.py -o /tmp/get-pip.py
      python3 /tmp/get-pip.py
    fi

    # 3) Improve packaging instruments and set up Ray
    python3 -m pip set up -U pip setuptools wheel
    python3 -m pip set up -U "ray(default)"

Hier finden Sie eine kurze Erläuterung jedes wichtigen YAML-Abschnitts.

cluster_name: Assigns a reputation to the cluster, permitting Ray to trace and handle 
it individually from others.

supplier:  Specifies which cloud to make use of (AWS right here), together with the area and 
availability zone for launching cases.

auth:  Defines how Ray connects to cases over SSH - the person identify and the 
non-public key used for authentication.

max_workers:  Units the utmost variety of employee nodes Ray can scale as much as when 
extra compute is required.

idle_timeout_minutes:  Tells Ray how lengthy to attend earlier than routinely terminating 
idle employee nodes.

available_node_types:  Describes the completely different node sorts (head and employees), their 
occasion sizes, AMI photos, and scaling limits.

head_node_type:  Identifies which of the node sorts acts because the cluster's controller
(the pinnacle node).

setup_commands:  Lists shell instructions that run as soon as on every node when it is first 
created, sometimes to put in software program or arrange the atmosphere.

Um die Clustererstellung zu starten, verwenden Sie diesen Ray-Befehl vom Terminal aus.

$ ray up -y ray_test.yaml

Ray wird seine Aufgabe erledigen und die gesamte erforderliche Infrastruktur erstellen. Nach ein paar Minuten sollten Sie in Ihrem Terminalfenster so etwas sehen.

...
...
...
Subsequent steps
  So as to add one other node to this Ray cluster, run
    ray begin --address='10.0.9.248:6379'

  To hook up with this Ray cluster:
    import ray
    ray.init()

  To submit a Ray job utilizing the Ray Jobs CLI:
    RAY_ADDRESS='http://10.0.9.248:8265' ray job submit --working-dir . -- python my_script.py

  See https://docs.ray.io/en/newest/cluster/running-applications/job-submission/index.html
  for extra info on submitting Ray jobs to the Ray cluster.

  To terminate the Ray runtime, run
    ray cease

  To view the standing of the cluster, use
    ray standing

  To observe and debug Ray, view the dashboard at
    10.0.9.248:8265

  If connection to the dashboard fails, test your firewall settings and community configuration.
Shared connection to 108.130.38.255 closed.
  New standing: up-to-date

Helpful instructions:
  To terminate the cluster:
    ray down /mnt/c/Customers/thoma/ray_test.yaml

  To retrieve the IP tackle of the cluster head:
    ray get-head-ip /mnt/c/Customers/thoma/ray_test.yaml

  To port-forward the cluster's Ray Dashboard to the native machine:
    ray dashboard /mnt/c/Customers/thoma/ray_test.yaml

  To submit a job to the cluster, port-forward the Ray Dashboard in one other terminal and run:
    ray job submit --address http://localhost:<dashboard-port> --working-dir . -- python my_script.py

  To hook up with a terminal on the cluster head for debugging:
    ray connect /mnt/c/Customers/thoma/ray_test.yaml

  To observe autoscaling:
    ray exec /mnt/c/Customers/thoma/ray_test.yaml 'tail -n 100 -f /tmp/ray/session_latest/logs/monitor*'

Ausführen eines Ray-Jobs auf einem Cluster

Zu diesem Zeitpunkt ist der Cluster erstellt und wir sind bereit, unseren Ray-Job an ihn zu senden. Um dem Cluster mehr Substanz zum Arbeiten zu geben, habe ich den Bereich für die Primzahlsuche in meinem Code von 10.000.000 auf 20.000.000 auf 10.000.000–60.000.000 erhöht. Auf meinem lokalen Desktop hat Ray dies in 18 Sekunden ausgeführt.

Ich habe kurz darauf gewartet, dass alle Cluster-Knoten vollständig initialisiert waren, und dann den Code mit diesem Befehl auf dem Cluster ausgeführt.

$  ray exec ray_test.yaml 'python3 ~/ray_test.py'

Hier ist meine Ausgabe.

(base) tom@tpr-desktop:/mnt/c/Customers/thoma$ ray exec ray_test2.yaml 'python3 ~/primes_ray.py'
2025-11-01 13:44:22,983 INFO util.py:389 -- setting max employees for head node sort to 0
Loaded cached supplier configuration
In case you expertise points with the cloud supplier, attempt re-running the command with --no-config-cache.
Fetched IP: 52.213.155.130
Warning: Completely added '52.213.155.130' (ED25519) to the record of identified hosts.
2025-11-01 13:44:26,469 INFO employee.py:1832 -- Connecting to current Ray cluster at tackle: 10.0.5.86:6379...
2025-11-01 13:44:26,477 INFO employee.py:2003 -- Linked to Ray cluster. View the dashboard at http://10.0.5.86:8265
nodes=6, CPUs~192, chunks=384
(autoscaler +2s) Tip: use `ray standing` to view detailed cluster standing. To disable these messages, set RAY_SCHEDULER_EVENTS=0.
(autoscaler +2s) No out there node sorts can fulfill useful resource requests {'CPU': 1.0}*160. Add appropriate node sorts to this cluster to resolve this problem.
complete=2897536, time=5.71s
Shared connection to 52.213.155.130 closed.

Wie Sie sehen, betrug die Zeit, die für die Ausführung des Clusters benötigt wurde, etwas mehr als 5 Sekunden. Somit führten fünf Employee-Knoten denselben Job in weniger als einem Drittel der Zeit aus, die auf meinem lokalen PC erforderlich warfare. Nicht zu schäbig.

Wenn Sie mit Ihrem Cluster fertig sind, führen Sie bitte den folgenden Ray-Befehl aus, um ihn zu zerstören.

$ ray down -y ray_test.yaml

Wie ich bereits erwähnt habe, sollten Sie Ihr Konto immer noch einmal überprüfen, um sicherzustellen, dass dieser Befehl wie erwartet funktioniert hat.

Zusammenfassung

Dieser Artikel ist der zweite Teil einer zweiteiligen Reihe und zeigt, wie Sie mithilfe der Ray-Bibliothek CPU-intensiven Python-Code auf cloudbasierten Clustern ausführen. Durch die Verteilung der Arbeitslast auf alle verfügbaren vCPUs stellt Ray sicher, dass unser Code schnelle Leistung und Laufzeiten liefert.

Ich habe beschrieben und gezeigt, wie man mithilfe einer YAML-Datei einen Cluster erstellt und wie man die Ray-Befehlszeilenschnittstelle verwendet, um Code zur Ausführung im Cluster zu übermitteln.

Unter Verwendung von AWS als Beispielplattform nahm ich Ray-Python-Code, der auf meinem lokalen PC ausgeführt wurde, und führte ihn – nahezu unverändert – auf einem EC2-Cluster mit 6 Knoten aus. Dies zeigte erhebliche Leistungsverbesserungen (3x) über die Nicht-Cluster-Laufzeit.

Abschließend zeigte ich, wie man mit dem Ray-Befehlszeilentool die von Ray erstellte AWS-Cluster-Infrastruktur abreißt.

Wenn Sie meinen ersten Artikel dieser Reihe noch nicht gelesen haben, klicken Sie auf den Hyperlink unten, um ihn anzusehen.