How to handle the tumbling window by java operator?
See below example in the document:
stream<int32 maxSalary, uint32 ageOfMaxSalary> Agg0 = Aggregate(Beat)
{
window
Beat : tumbling, count(5);
output
Agg0 : maxSalary = Max(salary), ageOfMaxSalary = ArgMax(salary, age);
}
I want to handle the tuples in the window when eviction. But where can I place the java operator? I do not think I can put operator after above code..
I did not find relating topics in info center....
Please help.
Thx
Your original question doesn't really make sense, the Aggregate operator is performing its own windowing using the windowing library, there is no mechanism to access the evicted tuples.
A Java Operator handles windowing by use of a window handler, not through its process method. Your operator registers an implementation of com.ibm.streams.operator.window.StreamWindowListener during initialization. Then your implementation will receive all window events (insertion, eviction etc.).
There are two sample Java operators that use windowing in the package com.ibm.streams.operator.samples.windows.
Windows can provide powerful capabilities but can be difficult to grasp initially. The following section covers windows thoroughly, please take a look at the examples provided to help you better understand windows.
Section 5.1.5 of the RedBook: "IBM InfoSphere Streams Assembling Continuous Insight in the Information Revolution" also covers Windows well. It can be downloaded here: http://www.redbooks.ibm.com/abstracts/SG247970.html?Open
I highly recommend this document.
Regarding Windowed operators in general, they allow the handling of a set of tuples. Streams can be consumed by operators either on a tuple-by-tuple basis or through windows that create logical groupings of tuples. If you want to handle single tuples use a non-windowed operator. Of course setting the trigger policy to count(1) will also cause processing to happen at the arrival or each tuple.
For handling tuples before they are evicted use a sliding window with a trigger policy set shorter than the eviction event. Sliding windows execute processing, including emitting an output stream based on the trigger policy.
You might find it helpful to think of a tumbling window as one where the eviction policy establishes the trigger policy.
I recommend you work with various operators using different window policies to help drive home the concepts.
Hi Stan,
thank you!
I am wondering if I want to handle the tuples in the window before it flushes. How can I do that?
In another case, if I am using a sliding window, when the window reaches its size, it will just evict one tuple (say, the oldest one), how can I handle the remain tuples in the window?
When the tunbling window evicts the the tuples it sends all of them to the output stream. When this stream is the input to the Java Primative it will receive all the tuples (22,42,34). I couldn't find good sample that demonstrated that so nodificed the JavaOperators sample that comes with Streams (see: <INSTALL>/samples/spl/feature/javaOperators). The attached SPL file is modified so the Java Primative DirectoryLister (source = DirectoryPropertyLister.java) recieves the evicted / flushed tuples from a SORT operator. Setup the sample program and replace the SPL code with the code in the one attached to this post. Give it a try to see that all the tuples are recieved in batches of five.
I see you've been active on the forum so perhaps you have figured this out already. Hope this helps.
Hi Stan,
sorry for misunderstanding you.
Let me take an example. For a tumbling window of size 3, its current status is (22,42,34). All the tuples in the window is going to evict after next tuple is inputting. If the next tuple is 30, then the tuples in the window will be (30), 22, 42 and 34 will be flushed. Hereby I want to handle these 3 figures by java operator. Which function should I choose? I thought process() can handle the new tuple 30, but I do not know how to handle the tuples (22,42,34) in the window.
Please let me know if I still can not make it clear :-)
I'm not understanding what you are trying to do. Would you please describe the processing you wish you implement - please include an input and output example. Your question about process() appears to be covered in the docs including how to implement an operator that handles information contained in more that one tuple.
Please see:
The section on persisting information across tuples:
A tuple that is passed in as a parameter to a process function can be used during the lifetime of the processfunction call. A pointer or a reference to the tuple should not be stored for use in a context other than that of the current process call. A tuple received from the process call can be safely submitted as long as the submit call is performed in the context of the process call. If the tuple is to be stored as part of the operator state and made available across process calls, then a copy should be made.
Note that the tuples can contain other Tuples and complex types and so is not like a standard row found in a common relational database. Some of the sample classes may be helpful:
A number of sample classes are provided that demonstrate operators implemented in Java:
- com.ibm.streams.operator.samples.operators.DecimalScaleSetter: Operator that submits a tuple for every input tuple, modifying a single decimal attribute by setting its scale (see java.math.BigDecimal.scale()).
- com.ibm.streams.operator.samples.patterns.TupleInTupleOut: An abstract pattern class that exposes filtering and transformation to a sub-class but hides the mechanics of tuple reception and submission. For each incoming tuple the sub-class's filter() method is called and if that returns false no further processing for that tuple is performed. If true is returned then transform() is called. Transform allows the sub-class to set the attributes of the outgoing tuple and simply return true to deliver the tuple to the output port.
- com.ibm.streams.operator.samples.operators.Regex: Operator that extends the TupleInTupleOut class to perform filtering using regular expression matching from the standard java.util.regex package.
Hello Stan,
thank you for the examples. They are comprehensive!
However, I think 033 can not solve my problem. In fact, I want to know how Java operator handle the current tuples in a tumbling or sliding window. But the process() function can only deal with with each incoming tuple, right?
You may find the following helpful, I know I do. I believe example 033 Java Primative Operator at Work will demonstrate what you what to do.
This package (SPL-Examples-For-Beginners.tar.gz) includes 50 different Streams Studio projects. These are simple examples that will help developers who are just beginning to wrap their minds around IBM Streams Processing Language (SPL).