WEBVTT 00:00.360 --> 00:01.240 Hi everyone. 00:01.240 --> 00:02.040 Welcome back. 00:02.080 --> 00:07.640 In this lecture we are going to explore one of the most exciting additions to the Java Stream API, 00:08.040 --> 00:10.760 which is the stream gatherers. 00:10.960 --> 00:20.760 Until Java 21 streams API were limited to predefined operations like map, filter and flatMap for intermediate 00:21.360 --> 00:22.320 operations. 00:22.520 --> 00:31.080 But now starting from Java 22, Java gives us a brand new way of building custom intermediate stream 00:31.120 --> 00:32.000 operations. 00:32.200 --> 00:34.960 And that's exactly what stream gatherers do. 00:35.440 --> 00:42.080 They let you manipulate the data flowing through a stream in ways that were previously impossible, 00:42.160 --> 00:49.560 like working with sliding windows, calculating running totals, or processing elements concurrently. 00:49.880 --> 00:52.000 We'll explore all of them in action. 00:52.000 --> 01:00.160 In the following lectures, you can find all the built in gatherers in the Java.util Stream Dot gatherers 01:00.160 --> 01:00.880 class. 01:00.920 --> 01:03.300 We'll explore that also in this lecture. 01:03.460 --> 01:09.460 So in this section we'll go step by step and explore what gatherers are and why we need them and how 01:09.460 --> 01:11.740 to use them effectively in your code. 01:11.780 --> 01:15.260 Let's talk about why do we need gatherers in the first place? 01:15.420 --> 01:18.580 The traditional stream model in Java is pull based. 01:18.780 --> 01:23.460 Each stage in the pipeline asks for data from the stage before it. 01:23.500 --> 01:30.300 For example, in this case, when you call collect or for each, the downstream operation pulls the 01:30.300 --> 01:31.980 data from the upstream. 01:31.980 --> 01:38.820 So in this case, the list of fruits is going to be passing the elements in the collection one by one. 01:39.060 --> 01:41.580 After we call the foreach operation. 01:41.580 --> 01:46.820 But gatherers flip this behavior, they make the stream push based. 01:46.820 --> 01:53.300 This means the upstream can now decide when and how to send elements downstream. 01:53.420 --> 01:59.220 It's not going to be the terminal operation that's going to be pulling the elements one by one. 01:59.460 --> 02:05.040 It's going to be the gatherer operation that's going to be pushing the elements downstream. 02:05.320 --> 02:08.520 Let me quickly show you an example to understand this behavior. 02:08.680 --> 02:15.720 So in this example here we have the gather operation which is an intermediate operation. 02:15.720 --> 02:21.000 And what it does is that here we have gatherers dot windows fixed of size three. 02:21.040 --> 02:29.080 This means that it's going to be gathering the elements of size of three elements, and then push those 02:29.080 --> 02:30.800 three elements downstream. 02:30.800 --> 02:32.960 So in this case to the for each operator. 02:33.000 --> 02:39.800 This gives us much more control, especially for stateful or batch oriented processing like grouping 02:39.800 --> 02:47.120 elements into windows, computing running totals or averages, or performing event streaming analytics. 02:47.280 --> 02:54.440 In short, gatherers are going to make the streams more powerful and expressive for real world data 02:54.440 --> 02:55.360 problems. 02:55.360 --> 02:57.760 We are going to be exploring some real time examples. 02:57.760 --> 03:02.280 Once we explore those examples, this concept will be much more clearer. 03:02.560 --> 03:08.020 So in this slide I'm going to be talking about the first gatherer, which is window fixed. 03:08.060 --> 03:11.060 This is exactly what we saw in the previous slide also. 03:11.220 --> 03:16.340 But in this slide I'm going to go through those windows fixed operator in detail. 03:16.340 --> 03:20.940 So this one groups the elements into fixed size batches. 03:20.940 --> 03:28.420 So basically when we call the gatherer operation it's going to be gathering the results from the original 03:28.420 --> 03:30.860 collection into elements of three. 03:31.260 --> 03:32.540 That's what you see over here. 03:32.540 --> 03:35.700 So here this is the movies is the actual collection. 03:35.700 --> 03:40.940 And then from the collection we do the gatherer dot window fixed of size three. 03:40.980 --> 03:46.900 This is going to be collecting the elements from the original collection into batches of three. 03:46.900 --> 03:50.260 So in this case let's say you have this movies collection of size ten. 03:50.420 --> 03:52.820 This is going to create four batches. 03:52.860 --> 03:54.620 Three batches will have three elements. 03:54.620 --> 03:57.060 And the last batch will have one element. 03:57.260 --> 04:02.900 And once that is collected then that list will be passed down to the for each operation. 04:02.900 --> 04:09.720 So in this case you can see that once the list of three elements of movie is collected, then it will 04:09.720 --> 04:12.240 push the batch to this for each operation. 04:12.240 --> 04:14.800 Now you have access to the whole list over here. 04:15.080 --> 04:18.680 Once we have this list, then we can actually use that list. 04:18.680 --> 04:22.640 And actually we are printing that in the console over here. 04:22.800 --> 04:31.280 So this is going to emit a window containing a list of type whatever that your actual stream is containing 04:31.280 --> 04:33.320 a subset of stream elements. 04:33.320 --> 04:35.240 So this is again going to create a stream. 04:35.240 --> 04:41.280 That's really important because you have a stream and that's going to be pushed down to the for each 04:41.320 --> 04:42.200 operation. 04:42.200 --> 04:46.600 So this is a concept of gathering it and pushing it downstream. 04:46.720 --> 04:48.320 I hope the explanation is clear. 04:48.360 --> 04:53.880 Now let's jump into the code and understand the gather API in detail and implement a basic example using 04:53.920 --> 04:57.000 the windows fixed gather operator. 04:57.000 --> 04:59.440 So I'm going to go back to the IntelliJ. 04:59.680 --> 05:03.600 So we are going to be working on this Stream Gatherers demo class. 05:03.760 --> 05:05.120 So that's what this one is. 05:05.120 --> 05:11.060 So I'm going to be searching for that gatherer API first gatherer. 05:11.100 --> 05:14.100 You can actually include this non-project items. 05:14.300 --> 05:14.700 There you go. 05:14.740 --> 05:16.420 This is the gatherer operator. 05:16.620 --> 05:18.380 So in this case you can see that. 05:18.380 --> 05:20.300 So this is gatherer interface. 05:20.300 --> 05:23.860 It accepts three elements actually the type of the input element. 05:23.860 --> 05:28.300 And then the potential mutable state type of gatherer operation. 05:28.300 --> 05:31.500 And then the output of the gatherer operation. 05:31.500 --> 05:36.820 Internally gatherers work through an integrator which receives this input streams. 05:36.820 --> 05:39.260 Basically you have the original stream. 05:39.260 --> 05:45.260 An integrator is the one which is going to be holding the actual intermediate state, and this is the 05:45.260 --> 05:48.900 one which will also decide when to emit the results downstream. 05:49.100 --> 05:54.460 So think of gatherers as like a smart transformers sitting in your stream pipeline. 05:54.740 --> 05:59.060 They can accumulate, hold and release elements based on the custom logic. 05:59.260 --> 06:05.020 This is what makes them more flexible than normal map or filter operations. 06:05.060 --> 06:09.730 Now let's switch back to the code and then code this very first function, which is going to be the 06:09.930 --> 06:12.450 demonstrate windows fixed function. 06:12.450 --> 06:19.530 And the data that we are going to be using for this particular example is the list of movie domain. 06:19.530 --> 06:22.210 Lets me go ahead and then show you what this domain is. 06:22.210 --> 06:24.650 So this domain has five properties. 06:24.690 --> 06:30.490 Number one is a title of the movie, the genre and the release date and the rating of the movie and 06:30.490 --> 06:31.370 the duration. 06:31.410 --> 06:32.330 And if we go back. 06:32.570 --> 06:35.930 So here we have this create sample movies function. 06:35.930 --> 06:39.770 So this function has a list of bunch of movies of different genres. 06:39.770 --> 06:44.450 So here you have drama movies, action movies science fiction romance. 06:44.570 --> 06:45.850 So these are the different genres. 06:45.850 --> 06:51.090 And each and every movie has different dates rating and duration. 06:51.090 --> 06:54.610 So the first thing that we are going to be exploring is the windows fixed. 06:54.770 --> 06:54.970 Okay. 06:55.010 --> 06:56.330 Let's go back to this one. 06:56.330 --> 06:59.090 And this is getting the input of list of movies. 06:59.130 --> 07:01.250 Let's go ahead and use that one. 07:01.250 --> 07:03.730 And then I'm going to be calling the stream function. 07:03.970 --> 07:10.010 So after the stream function we have access to the gather function call which is an operator. 07:10.010 --> 07:14.310 And this gather operator accepts a gatherer, actually. 07:14.350 --> 07:14.550 Okay. 07:14.590 --> 07:15.830 So that's what it accepts. 07:15.830 --> 07:19.190 If we go to the gather function again it accepts a gatherer. 07:19.190 --> 07:22.790 So we are going to be using some inbuilt gatherer to start with. 07:22.990 --> 07:26.870 We will build some custom gatherers as we go down in the future part of the course. 07:26.910 --> 07:29.070 Let's go back to the code here. 07:29.270 --> 07:34.390 And the gatherer that we are going to use is gatherer dot windows fixed. 07:34.390 --> 07:36.870 And then I'm going to give the value as three. 07:36.870 --> 07:38.310 So what this is going to return. 07:38.310 --> 07:43.710 So this is going to give you access to the stream of list of movies actually. 07:43.710 --> 07:49.790 So that's going to hold batches of three movies from the original movie collection. 07:49.790 --> 07:56.310 So this is going to be an intermediate operation creating a fixed windows of size threes okay. 07:56.350 --> 08:01.390 And what this is going to do, this is going to be passing it down to this for each function. 08:01.390 --> 08:03.870 And we are going to be accessing to that data. 08:03.870 --> 08:05.230 So I'm going to give a window. 08:05.390 --> 08:09.070 As you can see this is going to give you access to the list of movies. 08:09.070 --> 08:12.650 And after that I'm actually getting the access to the window. 08:12.650 --> 08:15.770 And then I'm actually printing each and every movie over here. 08:15.810 --> 08:17.890 Okay, let me put this on top. 08:18.090 --> 08:18.570 Okay. 08:18.610 --> 08:21.810 So this is our very first gatherer function. 08:21.850 --> 08:27.170 What this one does is that it's going to be acting on the original movies collection, and we're streaming 08:27.170 --> 08:27.610 on it. 08:27.650 --> 08:29.290 We are doing a gather over here. 08:29.530 --> 08:33.810 This gather is going to take care of gathering the movies of. 08:36.290 --> 08:39.170 List of three elements as a stream. 08:39.410 --> 08:42.810 And then when we call foreach, this gets pushed down to the. 08:43.010 --> 08:45.290 For each foreach is a terminal operation. 08:45.290 --> 08:48.090 And now this has access to the list of movies. 08:48.330 --> 08:51.490 And we are basically printing them in the console. 08:51.530 --> 08:53.530 Let me go ahead and execute this program. 08:53.610 --> 08:54.250 There you go. 08:54.610 --> 08:58.610 So now this is the actual movie list that you can see. 08:58.810 --> 09:02.890 And then we are actually gathering a list of three movies. 09:02.890 --> 09:08.330 In this case if we go to the demonstrate windows fixed, we are actually printing each window. 09:08.330 --> 09:11.370 And each window has access to list of three elements. 09:11.570 --> 09:17.350 And as you can see that it's going to be printing all the elements in the console. 09:17.510 --> 09:19.110 So this is the beauty of gatherers. 09:19.110 --> 09:23.670 You have the ability to batch the result and then pass it downstream. 09:23.670 --> 09:29.910 So if you think about it, this helps you create windows of exactly the specified size and windows do 09:29.910 --> 09:30.750 not overlap. 09:31.070 --> 09:33.270 And final window may contain fewer elements. 09:33.270 --> 09:37.910 If the stream size is not divisible by the window size, what does it mean actually? 09:37.910 --> 09:42.350 So in this case, let's say you have a collection of ten movies. 09:42.670 --> 09:48.350 Okay, now it will be broken down into four windows of size three, and the last window is going to 09:48.350 --> 09:49.990 be of size one. 09:50.030 --> 09:53.310 I think it's three windows actually not four three windows of size three. 09:53.310 --> 09:55.310 And the last window of size one. 09:55.350 --> 10:00.110 I hope you all have a pretty good idea about how gatherers work, and you can see the result one more 10:00.110 --> 10:00.390 time. 10:00.390 --> 10:04.550 So in the result we have windows fixed three fixed size windows. 10:04.550 --> 10:07.550 So we have window one has three elements. 10:07.550 --> 10:10.510 Window two has three elements window four and window. 10:10.510 --> 10:12.670 So totally there are four windows. 10:12.830 --> 10:16.090 Basically we have 12 movies in the original collection. 10:16.130 --> 10:17.650 I hope the explanation is clear. 10:17.770 --> 10:23.890 Let's go back to the slide and quickly highlight the differences between the streams and gatherers. 10:23.890 --> 10:29.770 So collectors are used at the end of the stream, right when you have processed everything, you are 10:29.770 --> 10:35.650 ready to combine all the results into one final output, for example a list or map. 10:35.650 --> 10:37.730 So this is how we do the collectors. 10:37.730 --> 10:43.730 Actually, in the case of gatherers, they are intermediate operations and they are push based and they 10:43.730 --> 10:50.570 have the ability to emit the intermediate results so it can collect the result and then emit the intermediate 10:50.570 --> 10:51.970 result to the downstream. 10:51.970 --> 10:58.210 But ultimately you need a downstream operation like in collector in order to receive the gathered results. 10:58.210 --> 11:01.890 So in this example, if you take a look at it, we have gathered the result here. 11:01.890 --> 11:07.170 But if you do not have a terminal operation, there is no way you can access this gathered result. 11:07.170 --> 11:13.050 So if you need fine grained control while elements are flowing through your pipeline gatherers or the 11:13.050 --> 11:14.210 tool to use. 11:14.250 --> 11:15.650 This marks the end of this lecture. 11:15.650 --> 11:16.690 Thank you for watching.