Enable the usage of generator functions in nodes

[idanov]

Description

Until now Kedro nodes were not very suitable for chunkwise processing of data. While nothing prevents a user to create a dataset loading the data as a generator, the only way one could process that data is by aggregating their result in memory and only then saving it.

This mode robs the chunkwise processing of data of its main use, namely minimising the memory footprint of the processing function. The changes in this PR make it possible to use generator functions in a Kedro pipeline and save the results by calling the dataset's save method separately for each chunk. Now the only thing a user needs to do in their function is:

for chunk in input:
    # process data chunk somehow
    yield processed_chunk

For taking advantage of this functionality, users need to create custom datasets which would operate in create-or-append-only mode, since the saving of the data. chunks happens over several invocations of the .save method. Chunk-wise data loading is already supported by existing Pandas datasets through the chunksize= load parameter, IIRC.

Note that the current implementation doesn't work with is_async data loading and saving.

Development notes

There are two main changes:

1. _outputs_to_dictionary in node.py
2. _run_node_sequential in runner.py

In 1., the main job is to convert a single generator to one or more generators, mapped to dataset names:

• when the node has one dataset as input, it returns the generator straight away in a dictionary with one key
• when the node has a list/tuple of outputs, it returns a dictionary with that list/tuple as keys, and the generator split into corresponding generators as values
• when the node has a dictionary mapping, it returns a dictionary with the generator split into a number of generators as values, mapped to the given names as keys

In order to manipulate the generators without invoking them, a combination of itertools and more_itertools functions are used.

In 2., the outputs are tested for being iterators, i.e. coming from generators. If that's the case, the generators are interleaved in a round-robin fashion and paired with their corresponding names. Example:

generator_a = (x for x in range(10))
generator_b = (x for x in range(10, 20))
outputs = {"a":  generator_a, "b": generator_b}
# after the processing the items stream will look like:
items = [("a", 0), ("b", 10), ("a", 1), ("b", 11), ...]

The new iterator will take the first element from each generator and pair them with the dataset name, then it will take the second element from each generator and pair them with their dataset name, and so on, until the generators finish. All generators are ensured to finish at the same time, because they originate from the same function generator.

This means that once the user function yields the data, each part will be saved to it's corresponding dataset, and then the function will continue processing the next chunk, yield the data, save it, then again process, yield, save, and so on.

Checklist

• Read the contributing guidelines
• Opened this PR as a 'Draft Pull Request' if it is work-in-progress
• Updated the documentation to reflect the code changes
• Added a description of this change in the RELEASE.md file
• Added tests to cover my changes

[jmholzer]

These changes are really cool, I love the efficient use of functions from itertools and more_itertools.

I left a few comments, all on code style, I couldn't find any logical problems though I haven't tested these changes manually yet.

[merelcht]

This was a really interesting PR to review! I learned a lot, but also still have a lot of questions 😁

I was also going to comment we should add docs for this and then saw you added an issue already 👍

[merelcht]

The code looks really good now! Great addition 👍

[jmholzer]

Nice! I really like this PR. The code is in a good state.

Would you like someone to manually test the changes before merging?

[idanov]

@jmholzer I think the manual test will be part of the documentation writing here: #2170