2
\$\begingroup\$

I have a batch job that contains a lot of items. I need to split these items into 4 categories - 1,2 category should be moved to a separate jobs as is; 3,4 categories job be additionally checked on validness and split into separate jobs - invalid items should be moved to a separate jobs.

Roughly diagram of the expected flow looks like this

enter image description here

For now I come up the prototype that looks like this

import java.util.*;
import java.util.stream.*;

public class MyClass {
    public static void main(String args[]) {
        Job job = new Job();
        job.items = Arrays.asList(
            new Item(ItemCategory.CAT_1, "1"),
            new Item(ItemCategory.CAT_2, "2"),
            new Item(ItemCategory.CAT_3, "3"),
            new Item(ItemCategory.CAT_1, "4"),
            new Item(ItemCategory.CAT_2, "5"),
            new Item(ItemCategory.CAT_3, "6"),
            new Item(ItemCategory.CAT_4, "7"),
            new Item(ItemCategory.CAT_4, "8")
        );

        /*
            [1:CAT_1, 4:CAT_1]
            [2:CAT_2, 5:CAT_2]
            [6:CAT_3]
            [8:CAT_4]
            [3:CAT_3, 7:CAT_4]
        */
        List<List<Item>> categorizedItems = job.items.parallelStream().collect(
            () -> Arrays.asList(new ArrayList<>(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>(), new ArrayList<>()), // last item is error aggregator
            (list, item) -> {
                switch (item.category) {
                    case CAT_1: list.get(0).add(item); break;
                    case CAT_2: list.get(1).add(item); break;
                    case CAT_3: {

                        if (isValid(item.id)) {
                            list.get(2).add(item);
                        } else {
                            list.get(4).add(item);
                        }

                        break;
                    }
                    case CAT_4: {

                        if (isValid(item.id)) {
                            list.get(3).add(item);
                        } else {
                            list.get(4).add(item);
                        }

                        break;
                    }
                }
            },
            (list1, list2) -> {
                list1.get(0).addAll(list2.get(0));
                list1.get(1).addAll(list2.get(1));
                list1.get(2).addAll(list2.get(2));
                list1.get(3).addAll(list2.get(3));
                list1.get(4).addAll(list2.get(4));
            }
        );
    }

    public static boolean isValid(String id) {
        return Integer.valueOf(id) % 2 == 0;
    }

    public static class Item {
        ItemCategory category;
        String id;

        Item(ItemCategory category, String id) {
            this.id = id;
            this.category = category;
        }

        @Override
        public String toString() {
            return id + ":" + category;
        }
    }

    public enum ItemCategory {
        CAT_1, CAT_2, CAT_3, CAT_4;
    }

    public static class Job {
        List<Item> items;
    }
}

Is it a good a idea to do everything in a single stream? Or for example it is better split into 4 categories first and then process 3,4 categories separately?

\$\endgroup\$
  • \$\begingroup\$ Re: “is it better split into 4 categories first?” Does the order of the items items that end up in category 5 matter? If you split the list into 4 groups, you’ve lost the relative ordering of invalid category 3 and invalid category 4 items. \$\endgroup\$ – AJNeufeld Feb 5 at 14:27
  • 1
    \$\begingroup\$ Out of curiosity, have you tried running this without the parallelStream? If you have a small data set, parallelism can be an overhead. \$\endgroup\$ – IEatBagels Feb 5 at 14:39
  • \$\begingroup\$ @AJNeufeld no it is not. category 5 is something akin Dead Letter Queue - as long as it contains right elements everything is ok. \$\endgroup\$ – lapots Feb 5 at 15:04
  • \$\begingroup\$ @IEatBagels I will have thousands of items in the list \$\endgroup\$ – lapots Feb 5 at 15:07
2
\$\begingroup\$

Your Collector is reinventing the wheel. There already exists a collector which partitions the collected items into groups: Collectors.groupingBy, which you could use something like:

.collect(
    Collectors.groupingBy(
        item -> (item.category == CAT_3 || item.category == CAT_4) && !isValid(item.id)
                     ? CAT_5 : item.category)
    )
)

Of course, this will return a Map<ItemCategory, List<Item>>. You can transform this back to a list of lists, if needed.

And of course, you would need to add the CAT_5 to the ItemCategory enum. Alternately, you could use null at the category 5 key, if you don't mind null as a key value, but be warned that it will make some people's skin crawl.


If the order of the items in the all of the groups (as opposed to just the Category 5 group) does not matter, then groupingByConcurrent will give better parallel stream performance.

\$\endgroup\$
  • \$\begingroup\$ But if I aggregate it into the map, then to process 3,4 category I will have to get them and store non valid items later into some other map or list? \$\endgroup\$ – lapots Feb 5 at 15:14
  • 1
    \$\begingroup\$ The classifier function, above, groups non-valid category 3,4 items into the map under category 5. Only valid category 3 items are stored in the map under category 3, and only valid category 4 items are stored in the map under category 4. \$\endgroup\$ – AJNeufeld Feb 5 at 16:09
  • \$\begingroup\$ did not know about groupingByConcurrent \$\endgroup\$ – lapots Feb 5 at 20:31

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.