Build Custom Flex Operators to Analyze Event Streams

Via the dropdown arrow of the POWER_OUTAGES element, click Modify > Edit Local Declaration(s). You can also press e.

Copy the CCL Script below.

DECLARE
  DICTIONARY ( string , msdate ) offtime ;
  TYPEOF ( POWER_OUTAGES ) outrec ;
  msdate offts ;
END

Paste it into the Edit Expression Value console, as seen below.

Now it’s time to add the block of CCL Script that will function as an event handler: it will be invoked on every event received on the DEVICE_EVENTS stream.

Note that a single flex operator can have any number of input streams and input windows, and must have an “ON” clause for each input.

In the POWER_OUTAGES Flex operator, expand the Methods compartment and double-click on ON DEVICE_EVENTS.

Copy the CCL Script shown here and paste it into the editor.

ON DEVICE_EVENTS {
  if (DEVICE_EVENTS.EVENT_VALUE = 'Power off') {
    offtime[DEVICE_EVENTS.MACHINEID] := DEVICE_EVENTS.EVENT_TIME;
  }
  if (DEVICE_EVENTS.EVENT_VALUE = 'Power on' AND not isnull(offtime[DEVICE_EVENTS.MACHINEID])) {
    offts := offtime[DEVICE_EVENTS.MACHINEID];
    output [
      MACHINEID = DEVICE_EVENTS.MACHINEID;
      POWER_OFF_TIME = offts;
      POWER_ON_TIME = DEVICE_EVENTS.EVENT_TIME;
      DURATION_MIN = cast(double,DEVICE_EVENTS.EVENT_TIME - offts)/60000000;];
  }
 };

Note: The msdate variable type is a timestamp with millisecond precision. The default format is YYYY-MMDDTHH:MM:SS:SSS. When an msdate is subtracted from another msdate, the resulting value is of type interval. The interval data type is in microseconds, hence the division by 60000000.

In this step you are going to do something a bit different. You’re going to add another Flex operator - but for a very different purpose. This Flex operator produces what you’ll call the “Dashboard” which has current status information for each machine. Now this is just a table - there’s no visualization - but it could easily be used to drive a dashboard.

The challenge is that your event stream contains different types of events on the same stream, with different uses of the “value” field. This is a common problem. But it prevents you from doing a simple aggregation or update, since one event can have a temperature reading in the value field and another event on the same stream can have “POWER_OFF” in the value field. So use CCL Script to create a new window called DASHBOARD that holds a single summary row for each machine. With each incoming event, the script examines the event and updates the relevant summary row.

• The local declare block creates a dictionary that holds the most recently received information for each machine. This dictionary starts out empty, but information is added/updated as events are received.
• The ON DEVICE_EVENTS method is invoked for each incoming event from the DEVICE_EVENTS stream. The method gets the previous set of information for the specific machine from the dictionary (if an entry for this machine exists in the dictionary). It then updates the information for this machine using the data contained in the event. Finally, it publishes an updated summary row for this machine and updates the dictionary with the current information.

With your project open in the CCL editor, copy the following CCL Script and paste it into your project (at the end).

CREATE FLEX DASHBOARD
IN DEVICE_EVENTS
OUT OUTPUT WINDOW DASHBOARD
  SCHEMA (
	MACHINEID string ,
	POWER_STATUS integer,
	CURR_TEMP float,
	MAX_TEMP float )
  PRIMARY KEY (MACHINEID)
  KEEP ALL
BEGIN
	DECLARE
	  typeof(DASHBOARD) outrec;
	  dictionary(string, typeof(DASHBOARD)) prev;
	END;
	ON DEVICE_EVENTS {
		if (not isnull(prev[DEVICE_EVENTS.MACHINEID])){
			outrec := prev[DEVICE_EVENTS.MACHINEID];
		}
		outrec.MACHINEID := DEVICE_EVENTS.MACHINEID;
		if(DEVICE_EVENTS.EVENT_NAME = 'POWER') {
			if (DEVICE_EVENTS.EVENT_VALUE = 'Power on') {
			outrec.POWER_STATUS := 0;}
			else {outrec.POWER_STATUS := 10;}
		}
		if (DEVICE_EVENTS.EVENT_NAME = 'TEMP') {
			outrec.CURR_TEMP := to_float(DEVICE_EVENTS.EVENT_VALUE);
			outrec.MAX_TEMP := to_float(DEVICE_EVENTS.MAX_TEMP);
		}
		output setOpcode(outrec,upsert);
		prev[DEVICE_EVENTS.MACHINEID] := outrec;
    } ;
END;

It should look like this:

View the complete project in the CCL editor. You should see something like this:

CREATE INPUT STREAM MACHINEDATA
SCHEMA (
	MACHINEID string ,
	EVENT_TIME msdate ,
	EVENT_NAME string ,
	EVENT_DESCRIPTION string ,
	EVENT_VALUE string ) ;

CREATE REFERENCE MACHINE_REF
    SCHEMA (
	MACHINEID string ,
	MACHINETYPE string ,
	MAX_TEMP decimal(4,2) ,
	MIN_TEMP decimal(4,2) ,
	LOCATION string ,
	TEMP_UNIT string )
	PRIMARY KEY ( MACHINEID )
	PROPERTIES service = 'hanadb' ,
	source = 'MACHINE_REF' ,
	sourceSchema = 'STREAMING' ;

/**@SIMPLEQUERY=FILTER*/
CREATE OUTPUT STREAM ACTIVITY_HIST
AS SELECT *
FROM MACHINEDATA
WHERE MACHINEDATA.EVENT_NAME = 'DOOR' ;

/**@SIMPLEQUERY=JOIN*/
CREATE OUTPUT STREAM DEVICE_EVENTS
AS SELECT
    MACHINEDATA.MACHINEID MACHINEID ,
	MACHINEDATA.EVENT_TIME EVENT_TIME ,
	MACHINEDATA.EVENT_NAME EVENT_NAME ,
	MACHINEDATA.EVENT_DESCRIPTION EVENT_DESCRIPTION ,
	MACHINEDATA.EVENT_VALUE EVENT_VALUE ,
	MACHINE_REF.MACHINETYPE MACHINETYPE ,
	MACHINE_REF.MAX_TEMP MAX_TEMP ,
	MACHINE_REF.MIN_TEMP MIN_TEMP ,
	MACHINE_REF.LOCATION LOCATION ,
	MACHINE_REF.TEMP_UNIT TEMP_UNIT
FROM MACHINEDATA INNER JOIN MACHINE_REF
ON MACHINEDATA.MACHINEID = MACHINE_REF.MACHINEID ;

/**@SIMPLEQUERY=AGGREGATE*/
CREATE OUTPUT WINDOW AVG_TEMP
PRIMARY KEY DEDUCED
KEEP ALL
AS SELECT
    DEVICE_EVENTS.MACHINEID MACHINEID ,
	LAST ( DEVICE_EVENTS.EVENT_TIME ) EVENT_TIME ,
	avg ( to_decimal(DEVICE_EVENTS.EVENT_VALUE, 4, 2) ) AVG_TEMP ,
	DEVICE_EVENTS.MAX_TEMP MAX_TEMP ,
	DEVICE_EVENTS.MIN_TEMP MIN_TEMP ,
	DEVICE_EVENTS.LOCATION LOCATION ,
	DEVICE_EVENTS.TEMP_UNIT TEMP_UNIT
FROM DEVICE_EVENTS KEEP 30 SEC
GROUP FILTER DEVICE_EVENTS.EVENT_NAME = 'TEMP'
GROUP BY DEVICE_EVENTS.MACHINEID ;

/**@SIMPLEQUERY=PATTERN*/
CREATE OUTPUT STREAM ALARM_POWER
AS SELECT
  A.MACHINEID MACHINEID ,
  A.EVENT_TIME EVENT_TIME ,
  A.LOCATION LOCATION ,
  'POWER' ALARM_TYPE ,
  'POWER Out for more than 20 seconds' ALARM_DESC
 FROM DEVICE_EVENTS A, DEVICE_EVENTS B
 MATCHING [ 20 SEC : A , ! B ]
 ON A.MACHINEID = B.MACHINEID
   AND 	A.EVENT_VALUE = 'Power off'
   AND B.EVENT_VALUE = 'Power on' ;

CREATE FLEX POWER_OUTAGES
IN DEVICE_EVENTS
OUT OUTPUT STREAM POWER_OUTAGES
SCHEMA (
	MACHINEID string ,
	POWER_OFF_TIME msdate ,
	POWER_ON_TIME msdate ,
	DURATION_MIN double )
BEGIN
  DECLARE
    dictionary(string, msdate) offtime;
    msdate offts;
  END;
  ON DEVICE_EVENTS {
  	if (DEVICE_EVENTS.EVENT_VALUE = 'Power off') {
  		offtime[DEVICE_EVENTS.MACHINEID] := DEVICE_EVENTS.EVENT_TIME;
  	}
  	if (DEVICE_EVENTS.EVENT_VALUE = 'Power on' AND not isnull(offtime[DEVICE_EVENTS.MACHINEID])) {
  		offts := offtime[DEVICE_EVENTS.MACHINEID];
  		output [
  			MACHINEID = DEVICE_EVENTS.MACHINEID;
  			POWER_OFF_TIME = offts;
  			POWER_ON_TIME = DEVICE_EVENTS.EVENT_TIME;
  			DURATION_MIN = cast(double,DEVICE_EVENTS.EVENT_TIME - offts)/60000000;];
  	}
  } ;
END;

/* Dashboard */
CREATE FLEX DASHBOARD
IN DEVICE_EVENTS
OUT OUTPUT WINDOW DASHBOARD
  SCHEMA (
	MACHINEID string ,
	POWER_STATUS integer,
	CURR_TEMP float,
	MAX_TEMP float )
  PRIMARY KEY (MACHINEID)
  KEEP ALL
BEGIN
	DECLARE
	  typeof(DASHBOARD) outrec;
	  dictionary(string, typeof(DASHBOARD)) prev;
	END;
	ON DEVICE_EVENTS {
		if (not isnull(prev[DEVICE_EVENTS.MACHINEID])){
			outrec := prev[DEVICE_EVENTS.MACHINEID];
		}
		outrec.MACHINEID := DEVICE_EVENTS.MACHINEID;
		if(DEVICE_EVENTS.EVENT_NAME = 'POWER') {
			if (DEVICE_EVENTS.EVENT_VALUE = 'Power on') {
			outrec.POWER_STATUS := 0;}
			else {outrec.POWER_STATUS := 10;}
		}
		if (DEVICE_EVENTS.EVENT_NAME = 'TEMP') {
			outrec.CURR_TEMP := to_float(DEVICE_EVENTS.EVENT_VALUE);
			outrec.MAX_TEMP := to_float(DEVICE_EVENTS.MAX_TEMP);
		}
		output setOpcode(outrec,upsert);
		prev[DEVICE_EVENTS.MACHINEID] := outrec;
    } ;
END;

CREATE OUTPUT WINDOW ALARM_TEMP
PRIMARY KEY DEDUCED
KEEP ALL
AS SELECT AVG_TEMP.MACHINEID MACHINEID ,
	AVG_TEMP.EVENT_TIME EVENT_TIME ,
	AVG_TEMP.LOCATION LOCATION ,
	'TEMP' ALARM_TYPE ,
	'Machine not maintaining temperature' ALARM_DESC
FROM AVG_TEMP
WHERE AVG_TEMP.AVG_TEMP > AVG_TEMP.MAX_TEMP ;

ATTACH OUTPUT ADAPTER HANA_Output1 TYPE hana_out TO ACTIVITY_HIST
PROPERTIES
	service = 'hanadb' ,
	sourceSchema = 'STREAMING' ,
	table = 'ACTIVITY_HIST' ;