Function chaining

Function chaining is a pattern where multiple functions are called on the same executor consecutively. Using the same lithops.FunctionExecutor object reference, multiple functions can be invoked. It increases the readability of the code and means less redundancy. This means we chain multiple functions together with the same element reference. It’s not necessary to attach the lithops.FunctionExecutor reference multiple times for each function call.

This pattern is especially useful when the output of one invocation is the input of another invocation. In this case, Lithops does not download the intermediate results to the local client; instead, the intermediate results are read directly by the next function.

It currently works with the Core API, and you can chain the map(), map_reduce(), wait(), and get_result() methods. Note that the returning value of one function must match the signature of the next function when chaining multiple map() calls. View the next examples:

Getting the result from a single map() call:

import lithops


def my_func1(x):
    return x * 2


iterdata = [1, 2, 3]

fexec = lithops.FunctionExecutor()
res = fexec.map(my_func1, iterdata).get_result()
print(res)

[2, 4, 6]

Chain multiple map() calls and get the final result:

import lithops


def my_func1(x):
    return x * 2, 5


def my_func2(x, y):
    return x + y


iterdata = [1, 2, 3]

fexec = lithops.FunctionExecutor()
res = fexec.map(my_func1, iterdata).map(my_func2).get_result()
print(res)

[7, 9, 11]

There is no limit in the number of map() calls that can be chained:

def my_func1(x):
    return x + 2, 5


def my_func2(x, y):
    return x + y, 5, 2


def my_func3(x, y, z):
    return x + y + z


iterdata = [1, 2, 3]

fexec = lithops.FunctionExecutor()
res = fexec.map(my_func1, iterdata).map(my_func2).map(my_func3).get_result()
print(res)

[15, 16, 17]

Alternatively, you can pass the futures generated in a map() or map_reduce() call to the iterdata parameter with the same effect. Not that in this case you will only get the results of the last map() execution. Results of intermediate map()s are never downloaded:

def my_func1(x):
    return x + 2, 5


def my_func2(x, y):
    return x + y, 5, 2


def my_func3(x, y, z):
    return x + y + z


iterdata = [1, 2, 3]

fexec = lithops.FunctionExecutor()
futures1 = fexec.map(my_func1, iterdata)
futures2 = fexec.map(my_func2, futures1)
futures3 = fexec.map(my_func3, futures2)
final_result = fexec.get_result()

print(final_result)

[15, 16, 17]

When the next job is invoked

By default a chained map() is invoked straight away, without waiting for the previous one. Each worker of the new job then blocks reading the result of its predecessor from storage, which is what keeps the intermediate results out of the client. The wait happens before the function starts, so it does not show up in worker_func_exec_time, but it is worker time you pay for: in a chain of two, where the first function takes 6 seconds, the second activation starts at the same instant as the first and stays idle for those 6 seconds.

Pass sync=True to have the client wait for the previous job before invoking the next one. The chain takes about the same wall-clock time, and no worker is billed for waiting:

import lithops


def my_func1(x):
    return x + 2, 5


def my_func2(x, y):
    return x + y


fexec = lithops.FunctionExecutor()
res = fexec.map(my_func1, [1, 2, 3]).map(my_func2, sync=True).get_result()
print(res)

Invoking without waiting is the better choice when the functions are short and the backend is slow to start a worker, since the two jobs overlap. Waiting is the better choice when a stage is long, when the fan-out is wide, or on a backend that bills for the whole activation.

What a chained function receives

A chained function is called with the result of the previous one and nothing else, so extra_args cannot be used together with chaining and raises a ValueError. Return the extra values from the previous function instead, or get the results of the chain and start a new job with them.

Any list or tuple of futures of a previous job works as the input of the next one, not only the list map() returned, so a slice or a comprehension chains the same way:

fexec = lithops.FunctionExecutor()
futures = fexec.map(my_func1, [1, 2, 3])
print(fexec.map(my_func2, futures[:2]).get_result())