更快的Python而无需重构您的代码

import numpy as npimport psutilimport rayimport scipy.signal num_cpus = psutil.cpu_count(logical=False) ray.init(num_cpus=num_cpus) @ray.remotedef 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)])

from multiprocessing import Poolimport numpy as npimport psutilimport 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 * [image], filters))

from collections import defaultdictimport numpy as npimport psutilimport ray num_cpus = psutil.cpu_count(logical=False) ray.init(num_cpus=num_cpus) @ray.remoteclass 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

from collections import defaultdictfrom multiprocessing import Poolimport numpy as npimport 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 = [set() for _ in range(4)] 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

import tensorflow as tf mnist = tf.keras.datasets.mnist.load_data()x_train, y_train = mnist[0]x_train = x_train / 255.0model = 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)

import psutilimport rayimport sysimport tensorflow as tf num_cpus = psutil.cpu_count(logical=False) ray.init(num_cpus=num_cpus) filename = '/tmp/model' @ray.remoteclass 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])

from multiprocessing import Poolimport psutilimport sysimport 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))

pip install numpy psutil ray scipy tensorflow

ray up config.yaml