你必須用對(duì)創(chuàng)建列表(filename, path)
data = [(img, img_dir) for img in images]
然后 map 將在單獨(dú)的進(jìn)程中運(yùn)行每一對(duì)。
但你必須args進(jìn)去start_pipeline
def start_pipeline(self, args):
print('ok starting')
filename, path = args
print('filename: {}\npath: {}'.format(filename, path))
return self.process()
你必須使用()創(chuàng)建類(lèi)的實(shí)例segmenter_class來(lái)使用start_pipeline
pool.map(segmenter_class().start_pipeline, data)
順便說(shuō)一句：在示例代碼中，我還返回了過(guò)程的結(jié)果。
import os
import multiprocessing
class SegmentationType(object):
DISPLAY_NAME = "invalid"
def __init__(self):
print('init')
def read_image(self):
print('read')
def write_image(self):
print('write')
def process(self):
# override in derived classes to perform an actual segmentation
pass
def start_pipeline(self, args):
print('ok starting')
filename, path = args
print('filename: {}\npath: {}'.format(filename, path))
return self.process()
class HSV_Segmenter(SegmentationType):
DISPLAY_NAME = 'HSV'
def process(self):
print('ok HSV')
return "result HSV"
class LabSegmenter(SegmentationType):
DISPLAY_NAME = 'LAB'
def process(self):
print('ok LAB')
return "result LAB"
if __name__ == '__main__':
procedure = 'hsv'
segmenter_class = {
'hsv': HSV_Segmenter,
'lab': LabSegmenter,
}.get(procedure)
images = ['01.png', '02.png', '03.png']
img_dir = 'C:/'
data = [(img, img_dir) for img in images]
pool = multiprocessing.Pool(3)
# example 1
results = pool.map(segmenter_class().start_pipeline, data)
print('Results:', results)
# example 2
for result in pool.map(segmenter_class().start_pipeline, data):
print('result:', result)
pool.terminate()
編輯：
您還可以創(chuàng)建獲取然后使用它的函數(shù)procedure-data這樣map每個(gè)進(jìn)程都會(huì)創(chuàng)建自己的實(shí)例，procedure或者您可以將不同的過(guò)程發(fā)送到不同的進(jìn)程。
import os
import multiprocessing
class SegmentationType(object):
DISPLAY_NAME = "invalid"
def __init__(self):
print('init')
def read_image(self):
print('read')
def write_image(self):
print('write')
def process(self):
# override in derived classes to perform an actual segmentation
pass
def start_pipeline(self, args):
print('ok starting')
filename, path = args
print('filename: {}\npath: {}'.format(filename, path))
return self.process()
class HSV_Segmenter(SegmentationType):
DISPLAY_NAME = 'HSV'
def process(self):
print('ok HSV')
return "result HSV"
class LabSegmenter(SegmentationType):
DISPLAY_NAME = 'LAB'
def process(self):
print('ok LAB')
return "result LAB"
segmenters = {
'hsv': HSV_Segmenter,
'lab': LabSegmenter,
}
def start_process(args):
procedure = args[0]
data = args[1:]
segmenter_class = segmenters.get(procedure)
result = segmenter_class().start_pipeline(data)
return result
if __name__ == '__main__':
procedure = 'hsv'
images = ['01.png', '02.png', '03.png']
img_dir = 'C:/'
data = [(procedure, img, img_dir) for img in images]
pool = multiprocessing.Pool(3)
# example 1
results = pool.map(start_process, data)
print('Results:', results)
# example 2
for result in pool.map(segmenter_class().start_pipeline, data):
print('result:', result)
pool.terminate()
不同程序的示例
if __name__ == '__main__':
images = ['01.png', '02.png', '03.png']
img_dir = 'C:/'
pool = multiprocessing.Pool(3)
data = [('hsv', img, img_dir) for img in images]
results = pool.map(start_process, data)
print('Results HSV:', results)
data = [('lab', img, img_dir) for img in images]
results = pool.map(start_process, data)
print('Results LAB:', results)
pool.terminate()
和一個(gè)一樣map()。有 6 個(gè)進(jìn)程要啟動(dòng)，Pool(3)因此它只會(huì)同時(shí)運(yùn)行 3 個(gè)進(jìn)程，當(dāng)它有空閑進(jìn)程時(shí)，map將從列表中獲取下一個(gè)值并運(yùn)行進(jìn)程。
if __name__ == '__main__':
images = ['01.png', '02.png', '03.png']
img_dir = 'C:/'
data_hsv = [('hsv', img, img_dir) for img in images]
data_lab = [('lab', img, img_dir) for img in images]
data = data_hsv + data_lab
pool = multiprocessing.Pool(3)
# example 1
results = pool.map(start_process, data)
print('Results:', results)
# example 2
for result in pool.map(start_process, data):
print('results:', result)
pool.terminate()
編輯：
它也適用于Ray
它只需要
from ray.util import multiprocessing
代替
import multiprocessing
我沒(méi)有用Dask、PySpark或Joblib測(cè)試它
編輯：
Joblib示例
from joblib import Parallel, delayed
class SegmentationType(object):
DISPLAY_NAME = "invalid"
def __init__(self):
print('init')
def read_image(self):
print('read')
def write_image(self):
print('write')
def process(self):
# override in derived classes to perform an actual segmentation
pass
def start_pipeline(self, args):
print('ok starting')
filename, path = args
print('filename: {}\npath: {}'.format(filename, path))
return self.process()
class HSV_Segmenter(SegmentationType):
DISPLAY_NAME = 'HSV'
def process(self):
print('ok HSV')
return "result HSV"
class LabSegmenter(SegmentationType):
DISPLAY_NAME = 'LAB'
def process(self):
print('ok LAB')
return "result LAB"
segmenters = {
'hsv': HSV_Segmenter,
'lab': LabSegmenter,
}
def start_process(args):
procedure = args[0]
data = args[1:]
segmenter_class = segmenters.get(procedure)
result = segmenter_class().start_pipeline(data)
return result
if __name__ == '__main__':
images = ['01.png', '02.png', '03.png']
img_dir = 'C:/'
data_hsv = [('hsv', img, img_dir) for img in images]
data_lab = [('lab', img, img_dir) for img in images]
data = data_hsv + data_lab
# --- version 1 ---
#pool = Parallel(n_jobs=3, backend='threading')
#pool = Parallel(n_jobs=3, backend='multiprocessing')
pool = Parallel(n_jobs=3)
# example 1
results = pool( delayed(start_process)(args) for args in data )
print('Results:', results)
# example 2
for result in pool( delayed(start_process)(args) for args in data ):
print('result:', result)
# --- version 2 ---
#with Parallel(n_jobs=3, backend='threading') as pool:
#with Parallel(n_jobs=3, backend='multiprocessing') as pool:
with Parallel(n_jobs=3) as pool:
# example 1
results = pool( delayed(start_process)(args) for args in data )
print('Results:', results)
# example 1
for result in pool( delayed(start_process)(args) for args in data ):
print('result:', result)
