0

对某些对象有一个复杂的计算系统。困难在于有些计算是分组计算。

这可以通过以下示例来证明：

from dask distributed import client

def load_data_from_db(id):
    # load some data
    ...
    return data

def task_a(data):
    # some calculations
    ...
    return result

def group_task(*args):
    # some calculations
    ...
    return result

def task_b(data, group_data):
    # some calculations
    ...
    return result

def task_c(data, task_a_result)
    # some calculations
    ...
    return result

ids = [1, 2]
dsk = {'id_{}'.format(i): id for i, id in enumerate(ids)}

dsk['data_0'] = (load_data_from_db, 'id_0')
dsk['data_1'] = (load_data_from_db, 'id_1')

dsk['task_a_result_0'] = (task_a, 'data_0')
dsk['task_a_result_1'] = (task_a, 'data_1')

dsk['group_result'] = (
    group_task,
    'data_0', 'task_a_result_0',
    'data_1', 'task_a_result_1')

dsk['task_b_result_0'] = (task_b, 'data_0', 'group_result')
dsk['task_b_result_1'] = (task_b, 'data_1', 'group_result')

dsk['task_c_result_0'] = (task_c, 'data_0', 'task_a_result_0')
dsk['task_c_result_1'] = (task_c, 'data_1', 'task_a_result_1')

client = Client(scheduler_address)
result = client.get(
    dsk,
    ['task_a_result_0',
     'task_b_result_0',
     'task_c_result_0',
     'task_a_result_1',
     'task_b_result_1',
     'task_c_result_1'])

对象列表数为数千个元素，任务数为数十个（包括几个组任务）。

使用这种图创建方法很难修改图（添加新任务、更改依赖项等）。对于这些上下文，是否有更有效的分布式计算方法？

添加

期货图是：

client = Client(scheduler_address)

ids = [1, 2]
data = client.map(load_data_from_db, ids)

result_a = client.map(task_a, data)

group_args = list(chain(*zip(data, result_a)))
result_group = client.submit(task_group, *group_args)

result_b = client.map(task_b, data, [result_group] * len(ids))

result_c = client.map(task_c, data, result_a)

result = client.gather(result_a + result_b + result_c)

在任务函数中，输入参数是Future实例，然后是 arg.result() 使用前。

4

1 回答 1

1

如果你想在计算过程中修改计算，那么我推荐期货接口。

于 2018-01-22T13:11:18.543 回答