Hi all, has someone used assets and iomanagers with chunks ? for example reading a huge file by chunk, process chunk by chunk and write them one by one but with somthing like coroutines or generators through iomanager ?

hi @Xavier BALESI! This sort of setup isn't currently possible purely through IOManagers -- the IOManager's load_input function can be a generator, but the generator will need to be fully exhausted before op execution starts (and I don't think you're able to yield more than a single value + metadata). I think you could potentially return a generator function from your io_manager, i.e.

def get_generator_function(context):
    def _fn():
        yield 1
    return _fn

def load_input(self, context):
    my_fn = self.get_generator_function(context)
    return my_fn

then invoke that generator in the op, but that only handles the input side. The output to be processed by the IOManager has to be emitted all at once (i.e.

yield Output(output_name="result", value=12345)

), so there's no direct support for chunked outputs. However, you could bring the output handling logic inside the body of the asset, and write chunks of your output to either the final location (meaning your IOManager's handle_output would essentially be a no-op), or a temporary file, which the IOManager would then copy over to a final location.

Hi @owen! Thank you very much for your response. I tried the "workaround" for the output with a temp file and it's OK. But I think it's not a very clean way for the long term. Do you think this feature could be added in the future ?

It sounds like something that would be possible to support (although probably somewhat involved), but it's not on the roadmap at the moment. I definitely encourage you to write up a github issue describing what the ideal behavior would look like for this use case!

done : ) https://github.com/dagster-io/dagster/issues/9243 thank you owen !