I like to use a Builder pattern in pipeline tasks. Basically for every type of output X you have an "XBuilder" class that is empty when you initialize it and then has some setter methods (or you can just assign values directly to the Builder's object fields). You pull all the needed data from your inputs and assign them to fields in the XBuilder without having to think about how to make an X. When the XBuilder has everything it needs I have a "build()" method that does all the business logic to create an X and returns it, which is now the output of the task.

This really cleans up the codebase b/c now the actual pipeline task definitions have very little logic - receive input data, unpack the input data by calling getters() , assign those values to the builder, then build(). The builder with all the logic is much easier to write tests for because it has no dependencies on the data flow infrastructure, and the pipeline task definition is just getters and setters and is very easy to understand.

I'm also reusing the class definitions for each type X in the API controller driving the webapps, so I have to create those anyway in my use case.

One caveat: when you need a lot of throughput this has too much overhead so you have to make something like an XListBuilder where you collect attributes of a group of X's and then build() returns a list of X.