Github user arunmahadevan commented on a diff in the pull request:
https://github.com/apache/storm/pull/1072#discussion_r57143268
--- Diff: storm-core/src/jvm/org/apache/storm/trident/Stream.java ---
@@ -594,19 +607,155 @@ public Stream aggregate(Fields inputFields,
ReducerAggregator agg, Fields functi
.aggregate(inputFields, agg, functionFields)
.chainEnd();
}
-
+
+ /**
+ * Returns a stream of tuples which are aggregated results of a
tumbling window with every {@code windowCount} of tuples.
+ *
+ * @param windowCount represents no of tuples in the window
+ * @param windowStoreFactory intermediary tuple store for storing
windowing tuples
+ * @param inputFields projected fields for aggregator
+ * @param aggregator aggregator to run on the window of tuples to
compute the result and emit to the stream.
+ * @param functionFields fields of values to emit with aggregation.
+ *
+ * @return
+ */
+ public Stream tumblingWindow(int windowCount, WindowsStoreFactory
windowStoreFactory,
+ Fields inputFields, Aggregator
aggregator, Fields functionFields) {
+ return window(TumblingCountWindow.of(windowCount),
windowStoreFactory, inputFields, aggregator, functionFields);
+ }
+
+ /**
+ * Returns a stream of tuples which are aggregated results of a
sliding window with every {@code windowCount} of tuples
+ * and slides the window with {@code slideCount}.
+ *
+ * @param windowCount represents tuples count of a window
+ * @param slideCount the number of tuples after which the window slides
+ * @param windowStoreFactory intermediary tuple store for storing
windowing tuples
+ * @param inputFields projected fields for aggregator
+ * @param aggregator aggregator to run on the window of tuples to
compute the result and emit to the stream.
+ * @param functionFields fields of values to emit with aggregation.
+ *
+ * @return
+ */
+ public Stream slidingWindow(int windowCount, int slideCount,
WindowsStoreFactory windowStoreFactory,
+ Fields inputFields, Aggregator
aggregator, Fields functionFields) {
+ return window(SlidingCountWindow.of(windowCount, slideCount),
windowStoreFactory, inputFields, aggregator, functionFields);
+ }
+
+ /**
+ * Returns a stream of tuples which are aggregated results of a window
tumbles at duration of {@code windowDuration}
+ *
+ * @param windowDuration represents tumbling window duration
configuration
+ * @param windowStoreFactory intermediary tuple store for storing
windowing tuples
+ * @param inputFields projected fields for aggregator
+ * @param aggregator aggregator to run on the window of tuples to
compute the result and emit to the stream.
+ * @param functionFields fields of values to emit with aggregation.
+ *
+ * @return
+ */
+ public Stream tumblingWindow(BaseWindowedBolt.Duration windowDuration,
WindowsStoreFactory windowStoreFactory,
+ Fields inputFields, Aggregator
aggregator, Fields functionFields) {
+ return window(TumblingDurationWindow.of(windowDuration),
windowStoreFactory, inputFields, aggregator, functionFields);
+ }
+
+ /**
+ * Returns a stream of tuples which are aggregated results of a window
which slides at duration of {@code slideDuration}
+ * and completes a window at {@code windowDuration}
+ *
+ * @param windowDuration represents window duration configuration
+ * @param slidingInterval the time duration after which the window
slides
+ * @param windowStoreFactory intermediary tuple store for storing
windowing tuples
+ * @param inputFields projected fields for aggregator
+ * @param aggregator aggregator to run on the window of tuples to
compute the result and emit to the stream.
+ * @param functionFields fields of values to emit with aggregation.
+ *
+ * @return
+ */
+ public Stream slidingWindow(BaseWindowedBolt.Duration windowDuration,
BaseWindowedBolt.Duration slidingInterval,
+ WindowsStoreFactory
windowStoreFactory, Fields inputFields, Aggregator aggregator, Fields
functionFields) {
+ return window(SlidingDurationWindow.of(windowDuration,
slidingInterval), windowStoreFactory, inputFields, aggregator, functionFields);
+ }
+
+ /**
+ * Returns a stream of aggregated results based on the given window
configuration which uses inmemory windowing tuple store.
+ *
+ * @param windowConfig window configuration like window length and
slide length.
+ * @param inputFields input fields
+ * @param aggregator aggregator to run on the window of tuples to
compute the result and emit to the stream.
+ * @param functionFields fields of values to emit with aggregation.
+ * @return
+ */
+ public Stream window(WindowConfig windowConfig, Fields inputFields,
Aggregator aggregator, Fields functionFields) {
+ // this store is used only for storing triggered aggregated
results but not tuples as storeTuplesInStore is set
+ // as false int he below call.
+ InMemoryWindowsStoreFactory inMemoryWindowsStoreFactory = new
InMemoryWindowsStoreFactory();
+ return window(windowConfig, inMemoryWindowsStoreFactory,
inputFields, aggregator, functionFields, false);
+ }
+
+ /**
+ * Returns stream of aggregated results based on the given window
configuration.
+ *
+ * @param windowConfig window configuration like window length and
slide length.
+ * @param windowStoreFactory intermediary tuple store for storing
tuples for windowing
+ * @param inputFields input fields
+ * @param aggregator aggregator to run on the window of tuples to
compute the result and emit to the stream.
+ * @param functionFields fields of values to emit with aggregation.
+ * @return
--- End diff --
`@return the new Stream`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
