Task/RowAggregatorTask.php
public function execute(ProcessState $state)
    {
        $input = $state->getInput();

        $aggregateBy = $this->getOption($state, 'aggregate_by');
        $aggregateColumns = $this->getOption($state, 'aggregate_columns');
        $aggregationKey = $this->getOption($state, 'aggregation_key');

        if (!array_key_exists($aggregateBy, $input)) {
            throw new InvalidProcessConfigurationException(
                "Array aggregator exception: missing column '{$aggregateBy}'"
            );
        }

        $inputAggregateBy = $input[$aggregateBy];

        if (!array_key_exists($inputAggregateBy, $this->result)) {
            $this->result[$inputAggregateBy] = $input;
            foreach ($aggregateColumns as $aggregateColumn) {
                if (array_key_exists($aggregateColumn, $this->result[$inputAggregateBy])) {
                    unset($this->result[$inputAggregateBy][$aggregateColumn]);
                }
            }
        }

        $inputAggregateColumns = [];
        foreach ($aggregateColumns as $aggregateColumn) {
            if (!array_key_exists($aggregateColumn, $input)) {
                throw new InvalidProcessConfigurationException(
                    "Array aggregator exception: missing column {$aggregateColumn}"
                );
            }
            $inputAggregateColumns[$aggregateColumn] = $input[$aggregateColumn];
        }
        $this->result[$inputAggregateBy][$aggregationKey][] = $inputAggregateColumns;
    }
public function execute(ProcessState $state)
    {
        $input = $state->getInput();

        $aggregateBy = $this->getOption($state, 'aggregate_by');
        $aggregateColumns = $this->getOption($state, 'aggregate_columns');
        $aggregationKey = $this->getOption($state, 'aggregation_key');

        if (!array_key_exists($aggregateBy, $input)) {
            throw new InvalidProcessConfigurationException(
                "Array aggregator exception: missing column '{$aggregateBy}'"
            );
        }

        $inputAggregateBy = $input[$aggregateBy];

        if (!array_key_exists($inputAggregateBy, $this->result)) {
            $this->result[$inputAggregateBy] = $input;
            foreach ($aggregateColumns as $aggregateColumn) {
                if (array_key_exists($aggregateColumn, $this->result[$inputAggregateBy])) {
                    unset($this->result[$inputAggregateBy][$aggregateColumn]);
                }
            }
        }

        $inputAggregateColumns = [];
        foreach ($aggregateColumns as $aggregateColumn) {
            if (!array_key_exists($aggregateColumn, $input)) {
                throw new InvalidProcessConfigurationException(
                    "Array aggregator exception: missing column {$aggregateColumn}"
                );
            }
            $inputAggregateColumns[$aggregateColumn] = $input[$aggregateColumn];
        }
        $this->result[$inputAggregateBy][$aggregationKey][] = $inputAggregateColumns;
    }
Store inputs and once everything has been received, pass to next task
Once an output has been generated this task is reset, and may wait for another loop

@param ProcessState $state

@throws \UnexpectedValueException
@throws \InvalidArgumentException
@throws \Symfony\Component\OptionsResolver\Exception\ExceptionInterface