We will now update the application in Kinesis Data Analytics to start counting trips per borough by updating the configuration. This shows how we can use Beam to count trips that cumulates by this new metric.
The Beam Java class that does is PartitionByBorough.java.
Go back to the Kinesis Data Analytics application and update the configuration. Select Confiuration and Scroll down to the Properties section you created earlier. Select Edit group for the group BeamApplicationProperties.
Update the property
true. Select update and wait a few minutes for the operator graph to update.
Changing this setting will change the internal application logic. Initially, the Beam application was using a global window to count all trip events in a 5 second window.
metrics = window .apply("Count globally", Combine .globally(Count.<TripEvent>combineFn()) .withoutDefaults() )
By adapting the parameter, the application will now first partition the trip events by their respective brough and then count the trip event per borough in the same 5 second window.
metrics = window .apply("Partition by borough", ParDo.of(new PartitionByBorough())) .apply("Count per borough", Count.perKey())
In both cases, the result is sent to Amazon CloudWatch for visualization.