A dynamic variable is used to store a tuple field value for later use. For example, your application may maintain a threshold value that is compared to a field value in each tuple processed. If the tuple field value is less than the threshold value, your application performs a set of operations whereas if the tuple field value is greater than, or equal to, the threshold value, your application performs a different set of operations.
If your application has no reason to change the threshold value while it runs, there is no reason to create a dynamic variable. You can hard-code the desired value into the application. But if you want to have the capability of changing the value while the application runs, you need the capabilities of a dynamic variable.
The value stored in a dynamic variable is obtained from a value in a tuple enqueued to the application. This tuple may be one of the tuples submitted to the application for processing, or it may be a control tuple, submitted to the application for the sole purpose of changing the application's configuration. The distinguishing characteristic of a dynamic variable is that its value changes while the application runs.
In a StreamSQL EventFlow application, the scope of a dynamic variable is the operator. Dynamic variables can be defined for all operators with the exception of Union, Split, Heartbeat, Metronome, and custom Java. To set the value of a dynamic variable, you must specify the stream on which the tuple with the new value can be found. In an EventFlow application, the only streams that can be uniquely identified are input and output streams. Therefore, the value of a dynamic variable is set from any tuple present on an input or output stream.
In a StreamSQL application, all streams are identifiable by name, so the tuple containing the field value used to update a dynamic variable can be obtained from any stream. Additionally, since in a StreamSQL application there is no equivalent to the EventFlow operator, the scope of a dynamic variable becomes the module (that is, the .ssql file) and not a single operator. As a result, in a StreamSQL application, the same dynamic variable may be used in multiple statements throughout the application whereas in an EventFlow application it would be necessary to define the same variable on multiple operators.
StreamSQL EventFlow support for Dynamic Variables is available in all current releases of StreamBase. StreamSQL support for Dynamic Variables is available in StreamBase 3.7 and subsequent releases.
Consider the following StreamSQL EventFlow application.
The predicate associated with the upper output port of the Filter operator compares a field in the tuple arriving on InputStream1 to a threshold value; for this example, assume that the fields being compared are integer types. If the tuple field value is less than the threshold value, the tuple is emitted to OutputStream1; otherwise the tuple is emitted to OutputStream2. Since there are no downstream operators to modify tuple field value, tuples at the output streams do not include a field that could be used to reset the threshold, and the threshold's value is simply hard-coded into the filter's predicate expression.
tuple_field_value < some_threshold_value
To gain the capabilities of a dynamic variable, you need to introduce another stream that contains the tuples that will be used to change the value stored in the variable, which leads to the following modified application where InputStream2 accepts tuples with a single field, newThreshold, which will be used to update the threshold value in the Filter operator.
Now you need to define a dynamic variable within the Filter operator and change the predicate expression so that it uses the dynamic variable. Select the Filter operator and on its Properties view, Dynamic Variables tab, add a row with the following entries to the Dynamic Variables: control.
Within the Filter operator, the dynamic variable has been assigned the name threshold. This variable will be updated from the newThreshold field in tuples arriving on InputStream2. Each time a tuple arrives on InputStream2, the value in its newThreshold field will be used to update the value stored in the dynamic variable threshold. Before any tuples arrive on InputStream2, the dynamic variable will be assigned an initial value of 5.
Now the running application will processes tuples arriving on InputStream1 and tuples satisfying the predicate expression tuple_field_value < threshold will be emitted on OutputStream1; all other tuples will be emitted on OutputStream2. To change the threshold value, submit a tuple to InputStream2.
An equivalent StreamSQL application may be easily written.
CREATE INPUT STREAM InputStream1 (
tuple_field_value int, ...
);
CREATE INPUT STREAM InputStream2 (
newThreshold int
);
CREATE OUTPUT STREAM OutputStream1;
CREATE OUTPUT STREAM OutputStream2;
DECLARE threshold int DEFAULT 5 UPDATE FROM
(SELECT newThreshold AS threshold
FROM InputStream2);
SELECT * FROM InputStream1
WHERE tuple_field_value < threshold
INTO OutputStream1
WHERE true
INTO OutputStream2;
The first four statements define the input and output streams; in the definition of InputStream1, the ellipsis indicates that there could be many other fields, but for the purposes of this discussion, only the tuple_field_value field is relevant. The last two statements illustrate how dynamic variables are used in a StreamSQL application.
The DECLARE statement defines the dynamic variable, and the SELECT statement uses the variable in its first WHERE clause. As with the EventFlow version of the application, the value assigned to the variable threshold will change as each tuple is enqueued onto InputStream2. Since a dynamic variable can only hold one value, the SELECT statement embedded within the DECLARE statement must select the single tuple field that will be used to reset the variable. The schema associated with InputStream2 could have additional fields that are used to reset other dynamic variables.
Now consider this application, where tuples not satisfying the tuple_field_value < threshold predicate are processed further in a Map operator.
The application uses the Map operator to examine the tuple_field_value in each tuple and, if a specific condition is met, resets the value stored in the dynamic variable threshold.
On the Map operator's Properties view, Output Settings tab, an additional field, newThreshold, is added to the tuple being processed and emitted. The actual logic used to set the value of this field is not important to this discussion except to note that this expression uses a dynamic variable, also named threshold. In the expression's logic, if the field tuple_field_value is greater than the current threshold value plus 5, then the threshold is increased by 5 (not to the current value of tuple_field_value)
The dynamic variable threshold is defined on the Map operator's Properties view, Dynamic Variables tab.
Although this variable has the same name as the variable used by the Filter operator, the application actually has two dynamic variables of the same name; scoping of dynamic variables in an EventFlow application is at the operator level.
The fact that the Map operator both defines a tuple field named newThreshold and then uses this field to set its own dynamic variable creates a processing loop, which will prevent the application from successfully type checking. StreamBase Studio detects this situation and now offers the ability to explicitly define the schema associated with the arc connecting the Map operator to OutputStream2.
Click on this arc and in its Properties view, Schema tab, Fields: control define the schema for tuples submitted to OutputStream2.
Now the application will type check and the value in the tuple field newThreshold can be used by both the Filter and Map operators to set their similarly named dynamic variables.
Again, writing the StreamSQL version of this application is straight-forward, however there are two significant differences.
- By explicitly defining the schema associated with OutputStream2 within its
CREATE OUTPUT STREAM statement, you can prepare the application for the loop situation and avoid the type check error.
- A single instance of the dynamic variable can be used in any statement within the application.
These differences will be apparent in the following StreamSQL application code.
CREATE INPUT STREAM InputStream1
(tuple_field_value int);
CREATE OUTPUT STREAM OutputStream1;
CREATE OUTPUT STREAM OutputStream2
(tuple_field_value int, newThreshold int);
CREATE STREAM LocalStream;
DECLARE threshold int DEFAULT 5 UPDATE FROM
(SELECT newThreshold AS threshold
FROM OutputStream2);
SELECT tuple_field_value
FROM InputStream1
WHERE tuple_field_value < threshold
INTO OutputStream1
WHERE true
INTO LocalStream;
SELECT tuple_field_value,
if(tuple_field_value>(threshold+5))
then (threshold+5)
else threshold
AS newThreshold
FROM LocalStream INTO OutputStream2;
Back to Top ^