Python의 multiprocessing 라이브러리는 다양한 응용 프로그램에 성공적으로 사용되어 왔지만, 이 블로그 게시물에서는 수치 데이터 처리, 상태를 유지하는 계산, 초기화가 비싼 계산을 포함한 여러 중요한 응용 프로그램 클래스 관련해서는 설명이 부족힙니다.
그 이유는 두 가지 주요 원인 때문입니다:

Numerical Data를 비효율적으로 처리함.
상태를 유지하는 계산에 대한 추상화가 부족함 (즉, 개별 “작업” 간에 변수를 공유할 수 없음).

Ray는 이러한 문제를 해결하기 위해 설계된 빠르고 간단한 분산 응용 프로그램 구축 및 실행 프레임워크입니다.
기본 개념에 대한 소개는 이 블로그 게시물을 참조하십시오.

Ray는 효율적인 데이터 처리를 위해 Apache Arrow를 활용하고, 분산 컴퓨팅을 위한 task 및 actor 추상화를 제공합니다.
이 블로그 게시물에서는 Python multiprocessing으로 쉽게 표현할 수 없는 세 가지 작업 부하를 벤치마크하고 Ray, Python multiprocessing, 그리고 단일 스레드 Python 코드를 비교합니다.
항상 최적화된 단일 스레드 코드와 비교하는 것이 중요합니다.

이 벤치마크에서 Ray는 단일 스레드 Python보다 10-30배 빠르고, multiprocessing보다 5-25배 빠르며, 큰 머신에서 이 둘 중 더 빠른 것보다 5-15배 빠릅니다.

48개의 물리적 코어를 가진 머신에서 Ray는 Python 멀티프로세싱보다 9배, 단일 스레드 Python보다 28배 빠릅니다. 오차 막대가 표시되었지만, 일부 경우에는 너무 작아서 보이지 않습니다. 이러한 수치를 재현할 수 있는 코드는 아래에서 확인할 수 있습니다. 작업 부하는 코어 수에 맞게 조정되므로, 더 많은 코어에서 더 많은 작업이 수행됩니다 (이 때문에 단일 스레드 Python은 더 많은 코어에서 더 오래 걸립니다).

벤치마크는 m5 인스턴스 유형(m5.large는 1개의 물리적 코어, m5.24xlarge는 48개의 물리적 코어)을 사용하여 EC2에서 실행되었습니다. 모든 벤치마크를 실행하는 코드는 여기에서 확인할 수 있습니다.
이 게시물에는 약간의 코드 조각이 포함되어 있습니다.

주요 차이점은 전체 벤치마크에는
1) 타이밍 및 출력 코드,
2) Ray 객체 저장소를 예열하는 코드,
3) 더 작은 머신에 벤치마크를 적응시키기 위한 코드

가 포함되어 있다는 점입니다.

벤치마크 1: Numerical Data

수치 데이터 많은 머신 러닝, 과학 컴퓨팅, 데이터 분석 작업 부하는 대규모 배열 데이터를 많이 사용합니다.
예를 들어, 배열은 큰 이미지나 데이터셋을 나타낼 수 있으며, 응용 프로그램은 여러 작업이 이미지를 분석하기를 원할 수 있습니다.
수치 데이터를 효율적으로 처리하는 것이 중요합니다.

아래 for 루프를 통해 각 패스를 수행하는 데 Ray는 0.84초, Python multiprocessing은 7.5초, 단일 스레드 Python은 24초가 걸립니다 (48개의 물리적 코어에서)
이 성능 차이는 Modin과 같은 라이브러리를 다른 라이브러리가 아닌 Ray 위에 구축할 수 있는 이유를 설명합니다.

코드는 다음과 같이 Ray로 작성됩니다.

import numpy as np
import psutil
import ray
import scipy.signal

num_cpus = psutil.cpu_count(logical=False)

ray.init(num_cpus=num_cpus)

@ray.remote
def f(image, random_filter):
    # Do some image processing.
    return scipy.signal.convolve2d(image, random_filter)[::5, ::5]

filters = [np.random.normal(size=(4, 4)) for _ in range(num_cpus)]

# Time the code below.

for _ in range(10):
    image = np.zeros((3000, 3000))
    image_id = ray.put(image)
    ray.get([f.remote(image_id, filters[i]) for i in range(num_cpus)])

위 코드는 다음과 같은 방식으로 동작합니다.

ray.put(image)를 호출하면 큰 배열이 공유 메모리에 저장됩니다.
또, 모든 워커 프로세스가 복사본을 생성하지 않고 이 배열에 접근할 수 있습니다.
이는 배열뿐만 아니라 배열을 포함하는 객체(ex, 배열 목록)에서도 동일하게 작동합니다.
워커가 f 작업을 실행할 때, 결과는 다시 공유 메모리에 저장됩니다.
스크립트가 ray.get([…])를 호출하면, 값을 역 직렬화 하거나 복사하지 않고도 공유 메모리를 기반으로 하는 numpy 배열을 생성합니다.
이러한 최적화는 Apache Arrow를 데이터 레이아웃 및 직렬화 형식으로 사용하고,
Plasma 공유 메모리 객체 저장소를 활용하는 Ray 덕분에 가능합니다.

코드는 Python 멀티프로세싱을 사용하여 다음과 같이 보입니다.

from multiprocessing import Pool
import numpy as np
import psutil
import scipy.signal

num_cpus = psutil.cpu_count(logical=False)

def f(args):
    image, random_filter = args
    # Do some image processing.
    return scipy.signal.convolve2d(image, random_filter)[::5, ::5]

pool = Pool(num_cpus)

filters = [np.random.normal(size=(4, 4)) for _ in range(num_cpus)]

# Time the code below.

for _ in range(10):
    image = np.zeros((3000, 3000))
    pool.map(f, zip(num_cpus * , filters))

여기서 차이점은 Python 멀티프로세싱이 프로세스 간에 큰 객체를 전달할 때 pickle을 사용하여 직렬화한다는 것입니다.
이 방법은 각 프로세스가 데이터를 자체적으로 복사해야 하며, 이로 인해 메모리 사용량이 크게 증가하고, 비싼 역직렬화에 대한 오버헤드가 발생합니다.
반면 Ray는 Apache Arrow 데이터 레이아웃을 사용하여 zero-copy serialization를 수행하고, Plasma Store를 사용하여 이러한 오버헤드를 피합니다.

벤치마크 2: 상태 저장 계산

많은 작은 작업 단위 간에 상당한 “상태”를 공유해야 하는 작업 부하도 Python 멀티프로세싱에 도전 과제가 되는 또 다른 유형의 작업 부하입니다.
이러한 패턴은 매우 흔하며, 여기에서는 장난감 스트림 처리 애플리케이션을 예로 들어 설명합니다.

48개의 물리적 코어를 가진 머신에서 Ray는 Python 멀티프로세싱보다 6배 빠르고, 단일 스레드 Python보다 17배 빠릅니다. Python 멀티프로세싱은 24개 이하의 코어에서는 단일 스레드 Python보다 성능이 뛰어나지 않습니다. 작업 부하는 코어 수에 따라 조정되므로, 더 많은 코어에서 더 많은 작업이 수행됩니다(이 때문에 단일 스레드 Python이 더 많은 코어에서 더 오래 걸리는 것입니다).

상태(state)는 종종 Python 클래스에 캡슐화되며, Ray는 클래스가 병렬 및 분산 환경에서 사용할 수 있도록 하는 actor 추상화를 제공합니다.
반면, Python 멀티프로세싱은 Python 클래스를 병렬화하는 자연스러운 방법을 제공하지 않기 때문에, 사용자는 종종 관련 상태를 map 호출 사이에 전달해야 합니다.
이 전략은 실제로 구현하기 어려울 수 있으며 (많은 Python 변수는 쉽게 직렬화되지 않습니다), 작동할 때도 느릴 수 있습니다.

아래는 병렬 작업을 사용하여 한 번에 하나의 문서를 처리하고, 각 단어의 접두사를 추출하며, 마지막에 가장 흔한 접두사를 반환하는 간단한 예제입니다.
접두사 카운트는 actor 상태에 저장되고 다른 작업에 의해 변경됩니다.

이 예제는 Ray로 3.2초, Python 멀티프로세싱으로 21초, 단일 스레드 Python으로 48개의 물리적 코어에서54초가 걸립니다.

Ray 버전 코드

Ray 버전은 다음과 같습니다.

from collections import defaultdict
import numpy as np
import psutil
import ray

num_cpus = psutil.cpu_count(logical=False)

ray.init(num_cpus=num_cpus)

@ray.remote
class StreamingPrefixCount(object):
    def __init__(self):
        self.prefix_count = defaultdict(int)
        self.popular_prefixes = set()

    def add_document(self, document):
        for word in document:
            for i in range(1, len(word)):
                prefix = word[:i]
                self.prefix_count[prefix] += 1
                if self.prefix_count[prefix] > 3:
                    self.popular_prefixes.add(prefix)

    def get_popular(self):
        return self.popular_prefixes

streaming_actors = [StreamingPrefixCount.remote() for _ in range(num_cpus)]

# Time the code below.

for i in range(num_cpus * 10):
    document = [np.random.bytes(20) for _ in range(10000)]
    streaming_actors[i % num_cpus].add_document.remote(document)

# Aggregate all of the results.
results = ray.get([actor.get_popular.remote() for actor in streaming_actors])
popular_prefixes = set()
for prefixes in results:
    popular_prefixes |= prefixes

Ray는 여기서 좋은 성능을 발휘하는데, 이는 Ray의 추상화가 해당 문제에 적합하기 때문입니다. 이 애플리케이션은 분산 환경에서 상태를 캡슐화하고 변형할 수 있는 방법이 필요하며, Ray의 액터(actor) 모델이 이를 충족시킵니다.

Multiprocessing 버전 코드

multiprocessing 버전은 다음과 같습니다.

from collections import defaultdict
from multiprocessing import Pool
import numpy as np
import psutil

num_cpus = psutil.cpu_count(logical=False)

def accumulate_prefixes(args):
    running_prefix_count, running_popular_prefixes, document = args
    for word in document:
        for i in range(1, len(word)):
            prefix = word[:i]
            running_prefix_count[prefix] += 1
            if running_prefix_count[prefix] > 3:
                running_popular_prefixes.add(prefix)
    return running_prefix_count, running_popular_prefixes

pool = Pool(num_cpus)

running_prefix_counts = [defaultdict(int) for _ in range(4)]
running_popular_prefixes = 

for i in range(10):
    documents = [[np.random.bytes(20) for _ in range(10000)]
                 for _ in range(num_cpus)]
    results = pool.map(
        accumulate_prefixes,
        zip(running_prefix_counts, running_popular_prefixes, documents))
    running_prefix_counts = [result[0] for result in results]
    running_popular_prefixes = [result[1] for result in results]

popular_prefixes = set()
for prefixes in running_popular_prefixes:
    popular_prefixes |= prefixes

여기서의 문제점은 pool.map이 stateless functions (상태없는 함수)를 실행한다는 점입니다.
즉, 한 pool.map 호출에서 생성된 변수를 다른 pool.map 호출에서 사용하려면, 첫 번째 호출에서 반환된 변수를 두 번째 호출로 전달해야 합니다.
작은 객체의 경우 이 접근 방식이 괜찮지만, 큰 중간 결과를 공유해야 할 때는 이를 전달하는 비용이 매우 큽니다(이 변수들이 스레드 간에 공유되는 경우에는 문제가 되지 않지만, 프로세스 경계를 넘어 공유될 때는 pickle과 같은 라이브러리를 사용해 바이트 문자열로 직렬화해야 하기 때문에 비용이 많이 듭니다).

이처럼 많은 상태를 전달해야 하기 때문에, multiprocessing 버전은 매우 어색하게 보이며, 결국 직렬 Python에 비해 작은 속도 향상만을 얻습니다.
실제로는 스트림 처리를 위해 Python multiprocessing을 사용하지 않을 것이기 때문에 이렇게 코드를 작성하지 않을 것입니다.
대신 전용 스트림 처리 프레임워크를 사용할 가능성이 높습니다. 이 예제는 Ray가 이러한 프레임워크나 애플리케이션을 구축하는 데 잘 맞는다는 것을 보여줍니다.

단, Python multiprocessing을 사용하는 방법은 여러 가지가 있습니다.
이 예제에서는 가장 유사한 API 비교를 제공하기 때문에 Pool.map과 비교합니다.
이 예제에서 더 나은 성능을 얻기 위해 별도의 프로세스를 시작하고, 그들 사이에 여러 multiprocessing queues를 설정할 수 있지만, 이는 복잡하고 취약한 설계를 초래할 수 있습니다.

벤치마크 3: 비용이 많이 드는 초기화

이전 예제와는 대조적으로, 많은 병렬 계산은 중간 계산이 작업 간에 공유될 필요는 없지만, 그럼에도 불구하고 이를 공유하는 것이 이득이 됩니다.
상태가 없는 계산도 초기화하는 데 비용이 많이 드는 상태를 공유함으로써 이득을 볼 수 있습니다.

다음은 디스크에서 저장된 신경망을 로드하여 병렬로 많은 이미지를 분류하려는 예제입니다.

48개의 물리적 코어가 있는 머신에서 Ray는 Python 멀티프로세싱보다 25배 빠르고 단일 스레드 Python보다 13배 빠릅니다. 이 예제에서는 Python 멀티프로세싱이 단일 스레드 Python보다 성능이 좋지 않습니다. 오류 막대가 표시되었지만, 일부 경우에는 너무 작아서 보이지 않을 수도 있습니다. 작업량은 코어 수에 맞게 조정되므로 더 많은 코어에서 더 많은 작업이 수행됩니다. 이 벤치마크에서 “직렬” Python 코드는 실제로 TensorFlow를 통해 여러 스레드를 사용합니다. Python 멀티프로세싱 코드의 변동성은 디스크에서 모델을 반복적으로 로드하는 변동성에서 비롯되며, 이는 다른 접근 방식에서는 필요하지 않습니다.

이 예제는 Ray로 5초, Python 멀티프로세싱으로 126초, 단일 스레드 Python으로 64초가 걸립니다(48개의 물리적 코어에서).
이 경우 단일 스레드 Python 버전은 실제로 단일 스레드가 아니며, TensorFlow를 통해 많은 코어를 사용하여 계산을 병렬화합니다.

예제 코드

다음과 같이 모델을 처음 생성했다고 가정해 보겠습니다.

import tensorflow as tf

mnist = tf.keras.datasets.mnist.load_data()
x_train, y_train = mnist[0]
x_train = x_train / 255.0
model = tf.keras.models.Sequential([
    tf.keras.layers.Flatten(input_shape=(28, 28)),
    tf.keras.layers.Dense(512, activation=tf.nn.relu),
    tf.keras.layers.Dropout(0.2),
    tf.keras.layers.Dense(10, activation=tf.nn.softmax)
])
model.compile(
    optimizer='adam',
    loss='sparse_categorical_crossentropy',
    metrics=['accuracy'])
# Train the model.
model.fit(x_train, y_train, epochs=1)
# Save the model to disk.
filename = '/tmp/model'
model.save(filename)

이제 모델을 로드하고 이를 사용하여 여러 이미지를 분류하고자 합니다.
애플리케이션에서는 이미지가 동시에 모두 사용 가능하지 않을 수 있으며, 이미지 분류가 데이터 로딩과 병렬로 이루어져야 할 수 있기 때문에 배치 단위로 이를 수행합니다.

Ray 버전 코드

Ray 버전은 다음과 같이 보입니다.

import psutil
import ray
import sys
import tensorflow as tf

num_cpus = psutil.cpu_count(logical=False)

ray.init(num_cpus=num_cpus)

filename = '/tmp/model'

@ray.remote
class Model(object):
    def __init__(self, i):
        # Pin the actor to a specific core if we are on Linux to prevent
        # contention between the different actors since TensorFlow uses
        # multiple threads.
        if sys.platform == 'linux':
            psutil.Process().cpu_affinity([i])
        # Load the model and some data.
        self.model = tf.keras.models.load_model(filename)
        mnist = tf.keras.datasets.mnist.load_data()
        self.x_test = mnist[1][0] / 255.0

    def evaluate_next_batch(self):
        # Note that we reuse the same data over and over, but in a
        # real application, the data would be different each time.
        return self.model.predict(self.x_test)

actors = [Model.remote(i) for i in range(num_cpus)]

# Time the code below.

# Parallelize the evaluation of some test data.
for j in range(10):
    results = ray.get([actor.evaluate_next_batch.remote() for actor in actors])

모델을 로드하는 속도가 느려서 한 번만 로드하려고 합니다.
Ray 버전에서는 액터의 생성자에서 모델을 한 번만 로드하여 이 비용을 분산시킵니다.
만약 모델을 GPU에 배치해야 한다면 초기화는 더욱 비용이 많이 듭니다.

Multiprocessing 버전은 매번 map 호출에서 모델을 다시 로드해야 하기 때문에 더 느립니다.
이는 매핑된 함수들이 상태가 없는 것으로 가정되기 때문입니다.

Multiprocessing 버전 코드

Multiprocessing 버전은 다음과 같이 보입니다.
일부 경우에는 multiprocessing.Pool의 initializer 인수를 사용하여 이를 달성할 수 있지만, 이는 각 프로세스에 대한 초기화가 동일할 때만 제한적으로 사용 가능합니다.
=====또한, 다른 프로세스가 다른 설정 기능(예: 다른 신경망 모델 로드)을 수행할 수 있도록 허용하지 않으며, 다른 작업을 다른 작업자에게 타겟팅할 수 없습니다.

from multiprocessing import Pool
import psutil
import sys
import tensorflow as tf

num_cpus = psutil.cpu_count(logical=False)

filename = '/tmp/model'

def evaluate_next_batch(i):
    # Pin the process to a specific core if we are on Linux to prevent
    # contention between the different processes since TensorFlow uses
    # multiple threads.
    if sys.platform == 'linux':
        psutil.Process().cpu_affinity([i])
    model = tf.keras.models.load_model(filename)
    mnist = tf.keras.datasets.mnist.load_data()
    x_test = mnist[1][0] / 255.0
    return model.predict(x_test)

pool = Pool(num_cpus)

for _ in range(10):
    pool.map(evaluate_next_batch, range(num_cpus))

우리가 위의 모든 예제에서 본 것은, Ray의 성능이 단순히 성능 최적화에서 오는 것이 아니라 주어진 작업에 적합한 추상화를 가지고 있다는 점입니다.
상태를 유지하는 계산은 많은 애플리케이션에서 중요하며, 상태를 유지하는 계산을 상태가 없는 추상화로 강제하려고 하면 비용이 듭니다.

벤치마크 실행하기

벤치마크를 실행하기 전에, 다음을 설치해야 합니다.

pip install numpy psutil ray scipy tensorflow

코드를 실행 시 위의 모든 숫자는 다음 스크립트를 실행하여 재현할 수 있습니다.

psutil 설치에 문제가 있다면 Anaconda Python을 사용해 보세요.
원래 벤치마크는 m5 인스턴스 타입(m5.large는 1개의 물리적 코어, m5.24xlarge는 48개의 물리적 코어)을 사용하여 EC2에서 실행되었습니다.
AWS 또는 GCP에서 올바른 구성으로 인스턴스를 시작하려면 Ray 클러스터 런처를 사용하고 다음 명령을 실행할 수 있습니다.

ray up config.yaml

m5.4xlarge 인스턴스를 시작하기 위한 예시 config.yaml입니다.

Ray에 대해 더 알아보기

이 블로그 포스트는 Ray와 Python 멀티프로세싱 간의 벤치마크에 중점을 두고 있지만,
두 라이브러리가 아주 유사하지는 않기 때문에 직접적인 비교는 어렵습니다. 다음과 같은 차이점이 있습니다.

Ray는 확장성을 염두에 두고 설계되어 있어 노트북과 클러스터에서 동일한 코드를 실행할 수 있습니다. 멀티프로세싱은 단일 머신에서만 실행됩니다.
Ray 작업은 머신 및 프로세스 실패로부터 자동으로 복구됩니다.
Ray는 언어에 구애받지 않도록 설계되었으며, Java에 대한 초기 지원을 제공합니다.

더 관련된 링크는 아래에 있습니다.

GitHub의 코드베이스.
Ray 문서.
Ray 포럼에서 질문하기.
Ray는 강화 학습, 하이퍼파라미터 튜닝, 모델 서비스 및 데이터 처리 확장을 위한 라이브러리를 포함합니다.
Ray는 AWS 및 GCP에서 클러스터를 시작하기 위한 클러스터 런처를 포함합니다.
Ray 웹페이지.

출처
https://towardsdatascience.com/10x-faster-parallel-python-without-python-multiprocessing-e5017c93cce1