Aggregate Operator Tuple Dimension Sample

Contents

Importing This Sample into StreamBase Studio
Sample Location
Running the Tuple-Based Aggregate Sample
How We Created the AggregateByTup Sample

This topic explains AggregateByTup.sbapp, the Aggregate operator tuple dimension sample.

There are two examples of the tuple-based aggregation in this sample. The first computes a moving average of shares based on a fixed number of trades. The second demonstrates how to create a sequence number using a window that does not close and which emits a value for each tuple received.

Importing This Sample into StreamBase Studio

In StreamBase Studio, import this sample with the following steps:

  • From the top menu, click File → Load StreamBase Sample.

  • Select operator from the Applications list.

  • Click OK.

StreamBase Studio creates a single project for the operator samples.

Sample Location

By default, the sample files are installed in:

  • On Windows: C:\Program Files\StreamBase Systems\StreamBase.n.m\sample\operator

  • On UNIX: /opt/streambase/sample/operator

When you load the sample into StreamBase Studio, Studio copies the sample project's files to your Studio workspace. StreamBase Systems recommends that you use the workspace copy of the sample, especially on UNIX, where you may not have write access to /opt/streambase. In the default installation, the path to this sample in your Studio workspace is: 

UNIX:       
  ~/streambase-studio-n.m-workspace/sample_operator
Windows XP:
  C:\Documents and Settings\username\My Documents\StreamBase Studio n.m Workspace\
      sample_operator
Windows Vista:
  C:\Users\username\Documents\StreamBase Studio n.m Workspace\
      sample_operator

Running the Tuple-Based Aggregate Sample

This sample demonstrates two uses of the tuple-based Aggregate operator. The first uses a tuple-based Aggregate operator to calculate a moving average of price for each of four successive trades of a given stock. It also ensures that the aggregation works for intermixed tuples for different stocks. The second provides a sequence number for input tuples, using a window which does not close and the emit feature to emit an output tuple for every input tuple.

Running AggregateByTup.sbapp in StreamBase Studio

  1. In the Package Explorer, double-click to open the AggregateByTup.sbapp application. Make sure the application is the currently active tab in the EventFlow Editor.

  2. Click the Run button. This opens the SB Test/Debug perspective and starts the application.

  3. In the Application Output view, select the AvgPricePSOut output stream. No output is displayed at this point, but the dequeuer is prepared to receive output. This view will eventually show the output of the application.

  4. In the Manual Input view, select the TradesIn input stream.

  5. Enter AMAT and 23 in the Symbol, and PricePerShare fields, respectively.

  6. Click Send Data, and observe that no output is displayed yet in the Application Output view.

  7. Enter AMAT and 24 in the Symbol, and PricePerShare fields, respectively.

  8. Click Send Data, and observe that no output is displayed yet in the Application Output view.

  9. Again enter AMAT and 24 in the Symbol, and PricePerShare fields, respectively.

  10. Click Send Data. There is still no output in the Application Output view.

  11. Enter AMAT and 25 in the Symbol, and PricePerShare fields, respectively.

  12. Click Send Data, and observe this line in the Application Output view:

    Symbol=AMAT, MovingAverage=24.0

    Tip

    If output data is too long to easily see in the Application Output table, click a row to display its field data in the Display Fields pane below the table.

  13. Enter AMAT and 20 in the Symbol, and PricePerShare fields, respectively.

  14. Click Send Data, and observe this line in the Application Output view:

    Symbol=AMAT, MovingAverage=23.25

  15. Enter AMAT and 21 in the Symbol, and PricePerShare fields, respectively.

  16. Click Send Data, and observe this line in the Application Output view:

    Symbol=AMAT, MovingAverage=22.5

  17. Enter AMAT and 21 in the Symbol, and PricePerShare fields, respectively.

  18. Click Send Data, and observe this line in the Application Output view:

    Symbol=AMAT, MovingAverage=21.75

  19. Enter AMAT and 22 in the Symbol, and PricePerShare fields, respectively.

  20. Click Send Data, and observe this line in the Application Output view:

    Symbol=AMAT, MovingAverage=21.0

  21. Enter INTC and 27 in the Symbol, and PricePerShare fields, respectively.

  22. Click Send Data, and observe no further output is displayed yet in the Application Output view.

  23. Enter INTC and 28 in the Symbol, and PricePerShare fields, respectively.

  24. Click Send Data, and observe no further output is displayed yet in the Application Output view:

  25. Enter AMAT and 22 in the Symbol, and PricePerShare fields, respectively.

  26. Click Send Data, and observe this line in the Application Output view:

    Symbol=AMAT, MovingAverage=21.5

  27. Enter INTC and 27 in the Symbol, and PricePerShare fields, respectively.

  28. Click Send Data, and observe no further output is displayed yet in the Application Output view:

  29. Enter AMAT and 23 in the Symbol, and PricePerShare fields, respectively.

  30. Click Send Data, and observe this line in the Application Output view:

    Symbol=AMAT, MovingAverage=22.0

  31. Enter INTC and 28 in the Symbol, and PricePerShare fields, respectively.

  32. Click Send Data, and observe this line in the Application Output view:

    Symbol=INTC, MovingAverage=27.5

  33. Next, view the sequence number.

  34. In the Application Output view, select the SeqOut output stream. No output is displayed at this point, but the dequeuer is prepared to receive output.

  35. Change from Manual Input to the Feed Simulations view.

  36. Right click on SeqFeed.sbfs and select Run Feed Simulation.

  37. Notice on output how sequence_number increases by 1.

  38. When done, press F9 or click the Stop Running Application button.

Running AggregateByTup.sbapp in Terminal Windows

This section describes how to run the sample in UNIX terminal windows or Windows command prompt windows. On Windows, be sure to use the StreamBase Command Prompt from the Start menu as described in the Test/Debug Guide, not the default command prompt.

  1. Open three terminal windows on UNIX, or three StreamBase Command Prompts on Windows. In each window, navigate to the directory where the sample is installed, or to your workspace copy of the sample, as described above.

  2. In window 1, type:

    sbd AggregateByTup.sbapp

    The window shows notice[StreamBaseServer] listening on port 10000.

  3. In window 2, type:

    sbc dequeue AvgPricePSOut

    No output is displayed at this point, but the dequeuer is prepared to receive output. This window will eventually show the output of the application.

  4. In window 3, type:

    sbc enqueue TradesIn

    The sbc command is now awaiting keyboard input. Then type:

    AMAT,23

    No output is displayed yet in the dequeue window.

  5. Type:

    AMAT,24

    No output is displayed yet in the dequeue window.

  6. Type:

    AMAT,24

    No output is displayed yet in the dequeue window.

  7. Type:

    AMAT,25

    Observe this line in the dequeue window:

    AMAT,24

  8. Type:

    AMAT,20

    Observe this line in the dequeue window:

    AMAT,23.25

  9. Type:

    AMAT,21

    Observe this line in the dequeue window:

    AMAT,22.5

  10. Type:

    AMAT,21

    Observe this line in the dequeue window:

    AMAT,21.750000

  11. Type:

    AMAT,22

    Observe this line in the dequeue window:

    AMAT,21

  12. Type:

    INTC,27

    No further output is displayed yet in the dequeue window.

  13. Type:

    INTC,28

    No further output is displayed yet in the dequeue window.

  14. Type:

    AMAT,22

    Observe this line in the dequeue window:

    AMAT,21.5

  15. Type:

    INTC,27

    No further output is displayed yet in the dequeue window.

  16. Type:

    AMAT,23

    Observe this line in the dequeue window:

    AMAT,22

  17. Type:

    INTC,28

    Observe this line in the dequeue window: INTC,27.5

  18. In window 3, press Ctrl-Z (Windows) or Ctrl-D (UNIX).

    The sbc enqueue command will exit.

  19. In window 2, press Ctrl-C.

    The sbc dequeue command will exit.

  20. In window 2, type:

    sbc dequeue SeqOut

  21. In window 3, type:

    sbfeedsim -a TradesIn2

    Observe the sequence numbers in the dequeue window

  22. Stop feedsim:

    In window 3, press Ctrl-C.

  23. In window 3, type the following command to terminate the server and dequeuer:

    sbadmin shutdown

How We Created the AggregateByTup Sample

  1. Launched StreamBase Studio.

  2. Created (or subsequently used) the sample_operator project.

  3. From the top menu, in the SB Authoring perspective, selected File → New → EventFlow Application. Selected the sample_operator project and entered AggregateByTup as the application name.

  4. Created an input stream for the moving average:

    1. Dragged an input stream from the palette to the EventFlow Editor.

    2. Clicked the stream on the EventFlow Editor, which invoked the Input Stream Properties dialog window.

    3. On the General tab, Name: TradesIn

    4. On the Edit Schema tab, added:

      • Field Name: Symbol, Type: string, Size: 5

      • Field Name: PricePerShare, Type: double

    5. Clicked OK.

  5. Created an Aggregate operator:

    1. Dragged an Aggregate operator from the palette to the EventFlow Editor.

    2. On the General tab, Name: AggregateTrades.

    3. Connect the TradesIn input stream to the AggregateTrades operator.

  6. Set up the AggregateTrades operator:

    1. On the Dimension tab, clicked the Add button. In the Edit Dimension dialog, added:

      Name: AggregateTradesDim

      Type: tuple.

      The buffer set up for each window will contain the specified number of tuples. When the buffer contains all the tuples required for the window, any desired calculations will take place, a tuple containing the desired results will be emitted, and the window will be closed. (See also the Aggregate Operator Field Dimension Sample.)

      Opening policy: Open per: Advance: 1.

      This indicates that a window should be open for a group of tuples. An Advance value of 1 advances the window by one tuple. A new window will be created as each tuple enters the system (containing that tuple), ready to accept the next n (where n = window size - 1) tuples. This means that windows overlap, and in this case, a given tuple may participate in up to four windows.

      Window size: Close and emit every 4

      The number of tuples in the buffer for this window.

      Emission policy: Selected "No intermediate emissions based on this dimension."

      "Emission policy" allows tuples to be emitted before the window closes. For example, one could emit a tuple every second during the 30-second window, rather than waiting for the window to close.

      Optional windows: Unchecked the Create partial windows checkbox.

      When set, this option creates partial windows which encompass the values that would have occurred before the arrival of the first tuple. For example, where Advance is less than Size, additional windows would be opened to include the first tuple; these windows would start before the Time in the first tuple.

      At this point our Edit Dimension dialog looks like this:

      Clicked OK.

    2. On the Aggregate Functions tab, unchecked the delta option, Output all input fields. Then added:

      Output Field Name: MovingAverage

      Expression: avg(PricePerShare)

      Calculates the average price per share of all the tuples in the window. For details about the available aggregate functions, see the StreamBase Expression Language and Functions topic in the Authoring Guide.

    3. On the Group Options tab, added:

      Output Field Name: Symbol

      Expression: Symbol

      Creates a window for each set of tuples whose value for the Symbol field is the same. Note that the Output Field Name need not be the same as the input field in the Expression. This is most useful when the Expression is more complicated than just a field value.

  7. Created an output stream:

    1. Dragged an output stream from the palette to the EventFlow Editor.

    2. On the General tab, Name: AvgPricePSOut

    3. Connected the AggregateTrades operator to the AvgPricePSOut output stream.

    That completed our work for the TradesIn stream.

Sequence Portion of the AggregateByTup Sample

To create the sequence portion of the AggregateByTup sample:

  1. Created a separate Input Stream for the sequence:

    1. Dragged an input stream from the palette to the EventFlow Editor.

    2. Clicked the stream on the EventFlow Editor, which invoked the Input Stream Properties dialog window.

    3. On the General tab, Name: TradesIn2

    4. On the Edit Schema tab, added:

      • Field Name: Symbol, Type: string, Size: 5

      • Field Name: PricePerShare, Type: double

    5. Clicked OK.

  2. Created an Aggregate operator:

    1. Dragged an Aggregate operator from the palette to the EventFlow Editor.

    2. On the General tab, Name: AggregateSeq.

    3. Connected the TradesIn2 Input Stream to the AggregateSeq operator.

  3. Set up the AggregateSeq operator:

    1. On the Dimension tab, clicked the Add button. In the Edit Dimension dialog, added:

      Name: AggregateSeqDim

      Type: tuple.

      The buffer set up for each window will contain the specified number of tuples. When the buffer contains all the tuples required for the window, any desired calculations will take place, a tuple containing the desired results will be emitted, and the window will be closed.

      Opening policy: Do not open window based on this dimension

      Window size: Do not close window based on this dimension

      That combination of settings creates a window that never closes.

      Emission policy: Intermediate intermission every 1

      "Emission policy" allows tuples to be emitted before the window closes. In this case, as every tuple arrives, one tuple is emitted with the sequence number.

      Optional windows: Unchecked the Open windows before first tuple checkbox.

      At this point our Edit Dimensions dialog for AggregateSeqDim looks like this:

    2. On the Aggregate Functions tab, unchecked the delta option, Output all input fields. Then added:

      Output Field Name: sequence_num

      Expression: count()

      As each tuple arrives in the aggregate operator, it will be counted. Since each tuple will cause a tuple to be emitted, the emitted tuple will contain a value one greater than the previous tuple. For details about the available aggregate functions, see the StreamBase Expression Language and Functions topic in the Authoring Guide.

    3. On the Group Options tab, did not add any groups.

      This example creates a sequence number that is global across all Symbols. If there were a group by on the Symbols field, each Symbol would have its own unique sequence.

  4. Created an output stream:

    1. Dragged an output stream from the palette to the EventFlow Editor.

    2. On the General tab, Name: SeqOut

    3. Connected the AggregateSeq operator to the SeqOut output stream.