Primer on Distributed Parallel Processing with Ray using KubeRay

In the early days of computing, applications handled tasks sequentially. As the scale grew with millions of users, this approach became impractical. Asynchronous processing allowed handling multiple tasks concurrently, but managing threads/processes on a single machine led to resource constraints and complexity.

This is where distributed parallel processing comes in. By spreading the workload across multiple machines, each dedicated to a portion of the task, it offers a scalable and efficient solution. If you have a function to process a large batch of files, you can divide the workload across multiple machines to process files concurrently instead of handling them sequentially on one machine. Additionally, it improves performance by leveraging combined resources and provides scalability and fault tolerance. As the demands increase, you can add more machines to increase available resources.

It is challenging to build and run distributed applications on scale, but there are several frameworks and tools to help you out. In this blog post, we'll examine one such open source distributed computing framework: Ray. We'll also look at KubeRay, a is a Distributed Parallel Processing framework that encapsulates all the benefits of distributed computing and solutions to challenges we discussed, such as fault tolerance, scalability, context management, communication, and so on. It is a

(.

Ray Core key methods

Below are some of the key methods within Ray Core that are commonly used:

• 
  

  ray.init() - Start Ray runtime and connect to the Ray cluster.
  
  

  
  
  CODE Kopieren

  import ray
  
  ray.init()
  
  

  
  
  


• 
  
  
  
  

  for more detailed information.

  
  
  

  Working with existing Python applications doesn’t require a lot of changes. The changes required would mainly be around the function or class that needs to be distributed naturally. You can add a decorator and convert it into tasks or actors. Let’s see an example of this.

  
  
  

  Converting a Python function into Ray Task
  
  

  
  
  
  
  

  CODE Kopieren

  # (Normal Python function)
  
  def square(x):
  
      return x * x
  
  
  
  # Usage
  
  results = []
  
  for i in range(4):
  
      result = square(i)
  
      results.append(result)
  
  print(results)
  
  
  
  # Output: [0, 1, 4, 9]
  
  
  
  
  
  # (Ray Implementation)
  
  # Define the square task.
  
  @ray.remote
  
  def square(x):
  
      return x * x
  
  
  
  # Launch four parallel square tasks.
  
  futures = [square.remote(i) for i in range(4)]
  
  # Retrieve results.
  
  print(ray.get(futures))
  
  # -> [0, 1, 4, 9]
  
  
  
  

  
  
  
  
  
  
  
  

  Converting a Python Class into Ray Actor
  
  

  
  
  
  
  

  CODE Kopieren

  # (Regular Python class)
  
  class Counter:
  
      def __init__(self):
  
          self.i = 0
  
  
  
      def get(self):
  
          return self.i
  
  
  
      def incr(self, value):
  
          self.i += value
  
  
  
  # Create an instance of the Counter class
  
  c = Counter()
  
  
  
  # Call the incr method on the instance
  
  for _ in range(10):
  
      c.incr(1)
  
  
  
  # Get the final state of the counter
  
  print(c.get())  # Output: 10
  
  
  
  # (Ray implementation in actor)
  
  # Define the Counter actor.
  
  @ray.remote
  
  class Counter:
  
      def __init__(self):
  
          self.i = 0
  
  
  
      def get(self):
  
          return self.i
  
  
  
      def incr(self, value):
  
          self.i += value
  
  
  
  # Create a Counter actor.
  
  c = Counter.remote()
  
  
  
  # Submit calls to the actor. These
  
  # calls run asynchronously but in
  
  # submission order on the remote actor
  
  # process.
  
  for _ in range(10):
  
      c.incr.remote(1)
  
  
  
  # Retrieve final actor state.
  
  print(ray.get(c.get.remote()))
  
  # -> 10
  
  

  
  
  
  
  
  
  
  

  Storing information in Ray Objects
  
  

  
  
  
  
  

  CODE Kopieren

  
  
  import numpy as np
  
  
  
  # (Regular Python function)
  
  # Define a function that sums the values in a matrix
  
  def sum_matrix(matrix):
  
      return np.sum(matrix)
  
  
  
  # Call the function with a literal argument value
  
  print(sum_matrix(np.ones((100, 100))))  # Output: 10000.0
  
  
  
  # Create a large array
  
  matrix = np.ones((1000, 1000))
  
  
  
  # Call the function with the large array
  
  print(sum_matrix(matrix))  # Output: 1000000.0
  
  
  
  
  
  # (Ray implementation of function)
  
  import numpy as np
  
  
  
  # Define a task that sums the values in a matrix.
  
  @ray.remote
  
  def sum_matrix(matrix):
  
      return np.sum(matrix)
  
  
  
  # Call the task with a literal argument value.
  
  print(ray.get(sum_matrix.remote(np.ones((100, 100)))))
  
  # -> 10000.0
  
  
  
  # Put a large array into the object store.
  
  matrix_ref = ray.put(np.ones((1000, 1000)))
  
  
  
  # Call the task with the object reference as argument.
  
  print(ray.get(sum_matrix.remote(matrix_ref)))
  
  # -> 1000000.0
  
  
  
  

  
  
  
  
  
  
  
  

  To learn more about its concept, head over to simplifies managing Ray clusters on Kubernetes by automating tasks such as deployment, scaling, and maintenance. It uses Kubernetes Custom Resource Definitions (CRDs) to manage Ray-specific resources.

  
  
  

  
  
  
  KubeRay CRDs
  

  
  
  

  It has three distinct CRDs.

  
  
  
  
  
  
  
  • 
    RayCluster: This CRD helps manage RayCluster's lifecycle and takes care of AutoScaling based on the configuration defined.
  
  
  • 
    RayJob: It is useful when there is a one-time job you want to run instead of keeping a standby RayCluster running all the time. It creates a RayCluster and submits the job when ready. Once the job is done, it deletes the RayCluster. This helps in automatically recycling the RayCluster.
  
  
  • 
    RayService: This also creates a RayCluster but deploys a RayServe application on it. This CRD makes it possible to do in-place updates to the application, providing zero-downtime upgrades and updates to ensure the high-availability of the application.
  
  
  
  
  

  
  
  
  Use-cases of KubeRay
  

  
  
  

  
  
  
  Deploying an on-demand model using RayService
  

  
  
  

  RayService allows you to deploy models on-demand in a Kubernetes environment. This can be particularly useful for applications like image generation or text extraction, where models are deployed only when needed.

  
  
  

  is an example of RayJob, which does Batch Inferencing.

  
  
  

  These are the use cases of KubeRay, enabling you to do more with the Kubernetes cluster. With the help of KubeRay, you can run mixed workloads on the same Kubernetes cluster and offload GPU-based workload scheduling to Ray.

  
  
  

  
  
  
  Conclusion
  

  
  
  

  Distributed parallel processing offers a scalable solution for handling large-scale, resource-intensive tasks. Ray simplifies the complexities of building distributed applications, while KubeRay integrates Ray with Kubernetes for seamless deployment and scaling. This combination enhances performance, scalability, and fault tolerance, making it ideal for web crawling, data analytics, and machine learning tasks. By leveraging Ray and KubeRay, you can efficiently manage distributed computing, meeting the demands of today's data-driven world with ease.

  
  
  

  Not only that, but as our compute resource types are changing from CPU to GPU-based, it becomes important to have efficient and scalable cloud infrastructure for all sorts of applications, whether it be AI or large data processing. For that, you can bring in

  Vollständiger Original-Bericht

  Ausführliche Details, Code-Beispiele & Hersteller-Stellungnahme auf dev.to.

  ↗ Original-Artikel auf dev.to lesen