Share via

StreamInsight and reference data (lists, databases, etc)

  • Article

Using reference data in StreamInsight is a very common scenarios; some examples would be:

  • Monitoring process control event streams for values that exceed a given threshold (for example, a valve’s pressure exceeding a certain safety threshold).
  • Enriching utility usage information in a smart metering scenario with user and geography information (for example, to allow visualization of power usage by geography).

This reference data is commonly held in some form of repository, such as SQL Server, a process historian’s metadata store (e.g. OSIsoft’s PI Asset Framework), which then needs to be integrated into a StreamInsight query.

In my previous post, I showed how to use the TimeImport capability on streams to synchronize a slow moving reference stream with a fast moving data stream.  I did not, however, walk through the process of creating a reference stream from a relatively static source (such as a database table or file).  I intend to now correct that minor, and somewhat deliberate oversight Smile

The project referenced in this blog post, along with the source data files is available here.

Reference data can be integrated into StreamInsight in a couple of different ways, each with their own pros and cons:

Technique Pros Cons
  • User-defined lookup function
  • Relatively simple to implement
  • Not suitable for large dynamic data sets or data that changes over time
  • Cumbersome to import more than one data element
  • Cannot take advantage of or incorporate stream data
  • Reference data stream
  • Seamless integration with StreamInsight query experience
  • Very performant and flexible
  • Additional work required to import and maintain the stream

Let’s look at an example of each technique, step by step. 

The Scenario

For this illustration, let’s consider the following scenario:

  • Stream of data events, each containing a user ID, an activity code and a status (user 7, logged in, success).
  • Set of user metadata, contained in a file (for the purposes of illustration – a database would be a more likely source).  This metadata contains information such as user ID, user name, etc.
  • Want to annotate the data events with the user metadata.

We’ll use the following data sets:

Data Events

Start Time End Time User Id Activity Code Status

6/25/2009 0:00:00

   1 Logon Success

6/25/2009 0:00:01

   2 Logon Fail

6/25/2009 0:00:05

   1 Browse Success

6/25/2009 0:00:06

   2 Logon Success

Or, in the .csv format that we’ll use:

 StartTime,EndTime,UserId,ActivityCode,Status6/25/2009 0:00:00,,1,Logon,Success6/25/2009 0:00:01,,2,Logon,Fail6/25/2009 0:00:05,,1,Browse,Success6/25/2009 0:00:06,,2,Logon,Success

Reference Events

Start Time End Time User Id User Name Location

6/25/2009 0:00:00

   1 Fred Jones Seattle

6/25/2009 0:00:01

   2 Bob Murphy Portland

Or, in the .csv format that we’ll use:

 StartTime,EndTime,UserId,UserName,Location6/24/2009 0:00:00,,1,Fred Jones,Seattle6/24/2009 0:00:01,,2,Bob Murphy,Portland

User-Defined Lookup Function

With a user defined lookup function, we need to load the list of metadata into an associative array (i.e. a hashtable), and create a lookup function that allows us to grab user metadata from the dictionary on demand.

Note: Why did I use the hard coded dictionary rather than reading from a file?  I wanted to illustrate the technique of a truly static data source.  If you want to see the code that parses a .csv file into a dictionary object suitable for using in a lookup function, this is covered in the appendix section of the blog).

Note: Why didn’t I simply join the dictionary with the data stream?  StreamInsight needs to join streams both relationally and temporally.  Since a dictionary isn’t a stream (and has no concept of time, it’s not possible to directly join.  Converting a reference source into a stream and joining is the other technique covered in the next section.

Code Snippet

  1. static Query CreateUserDefinedFunction(Application cepApp)
  2. {
  3.     var dataStream = CepStream<DataEvent>.Create("dataStream",
  4.         typeof(TextFileReaderFactory), new TextFileReaderConfig()
  5.         {
  6.                 InputFileName = "dataEvents.csv",
  7.                 CtiFrequency = 1,
  8.                 CultureName = CultureInfo.InvariantCulture.Name,
  9.                 Delimiter = ','                       
  10.         },  EventShape.Point);
  11.  
  12.     // Populate the reference data statically
  13.     refData.Add(1, new ReferenceData()
  14.         { UserId = 1, UserName = "Fred Jones", Location = "Seattle" } );
  15.     refData.Add(2, new ReferenceData()
  16.         { UserId = 2, UserName = "Bob Murphy", Location = "Portland" } );
  17.             
  18.     // Create a stream with the user name added to the query
  19.     var joinedStream = from e in dataStream
  20.         select new
  21.         {
  22.             UserId = e.UserId,
  23.             Activity = e.ActivityCode,
  24.             Status = e.Status,
  25.             // Use a static lookup function to retrieve the name
  26.             Name = LookupName(e.UserId),
  27.             // Use a static lookup function to retrieve the location
  28.             Location = LookupLocation(e.UserId)
  29.         };
  30.  
  31.     // Attach a trace output adapter to the stream
  32.     var query = joinedStream.ToQuery(cepApp, "outputLookup", "",
  33.         typeof(TracerFactory), new TracerConfig()
  34.         {
  35.             DisplayCtiEvents = false,
  36.             SingleLine = false,
  37.             TraceName = "REF",
  38.             TracerKind = TracerKind.Console
  39.         }, EventShape.Point, StreamEventOrder.FullyOrdered);
  40.  
  41.     return query;
  42. }
  43.  
  44. public static string LookupName(int userId)
  45. {
  46.     if (refData.ContainsKey(userId))
  47.         return refData[userId].UserName;
  48.     else
  49.         return "Unknown";
  50. }
  51.  
  52. public static string LookupLocation(int userId)
  53. {
  54.     if (refData.ContainsKey(userId))
  55.         return refData[userId].Location;
  56.     else
  57.         return "Unknown";
  58. }

Upon running this function with the supplied data files, we observe the merged output:

 Press <enter> to close applicationREF,Point,12:00:00.000        Activity = Logon        Location = Seattle        Name = Fred Jones        Status = Success        UserId = 1REF,Point,12:00:01.000        Activity = Logon        Location = Portland        Name = Bob Murphy        Status = Fail        UserId = 2REF,Point,12:00:05.000        Activity = Browse        Location = Seattle        Name = Fred Jones        Status = Success        UserId = 1REF,Point,12:00:06.000        Activity = Logon        Location = Portland        Name = Bob Murphy        Status = Success        UserId = 2

Reference Data Stream (static data)

Using the more robust reference data stream approach, we:

  • Use a TextFileInputFactory to import the contents of the text file
  • Convert the series of point events into “infinite” signals.
  • Join the reference stream with the data stream.

Note: this addresses conditions where the reference data DOES NOT CHANGE.  This is not a very real-world scenario – we’ll cover the additions required to incorporate changing data in the next blog post, showing how to pull data from SQL Server. 

Note: I don’t use the CTI import technique in this example.  This is due to the text file input adapter sending an “infinite time” CTI when it reaches the end of the file.  If we were using a continuous or changing reference data source (such as SQL Server), I’d need to employ that technique.

Code Snippet

  1. static Query CreateThresholdStream(Application cepApp)
  2. {
  3.     // Pull the data events from the dataEvents.csv file
  4.     var dataStream = CepStream<DataEvent>.Create("dataStream",
  5.         typeof(TextFileReaderFactory), new TextFileReaderConfig()
  6.         {
  7.                 InputFileName = "dataEvents.csv",
  8.             CtiFrequency = 1,
  9.             CultureName = CultureInfo.InvariantCulture.Name,
  10.             Delimiter = ','                     
  11.         }, EventShape.Point);
  12.  
  13.     // Create a reference stream using the datastream as a time reference
  14.     var refStream = CepStream<ReferenceData>.Create("refStream",
  15.         typeof(TextFileReaderFactory), new TextFileReaderConfig()
  16.         {
  17.             InputFileName = "refEvents.csv",
  18.             CtiFrequency = 1,
  19.             CultureName = CultureInfo.InvariantCulture.Name,
  20.             Delimiter = ','
  21.         }, EventShape.Point);
  22.  
  23.     // Stretch the ref stream points events out to infinity (i.e. must have
  24.     // relational and temporal matches to join, and for this example, the data
  25.     // doesn't change.
  26.     var referenceEvents = from e in refStream.AlterEventDuration(
  27.                                 e => TimeSpan.MaxValue)
  28.                             select e;
  29.  
  30.     // Join the two streams
  31.     var joinedStream = from e1 in dataStream
  32.                         join e2 in referenceEvents
  33.         on e1.UserId equals e2.UserId
  34.         select new
  35.         {
  36.             UserId = e1.UserId,
  37.             Activity = e1.ActivityCode,
  38.             Status = e1.Status,
  39.             // Use the reference stream to retrieve the name
  40.             Name = e2.UserName,
  41.             // Use the reference stream to retrieve the location
  42.             Location = e2.Location
  43.         };
  44.  
  45.     var query = joinedStream.ToQuery(cepApp, "outputStream", "",
  46.         typeof(TracerFactory), new TracerConfig()
  47.         {
  48.             DisplayCtiEvents = false,
  49.             SingleLine = false,
  50.             TraceName = "STREAM",
  51.             TracerKind = TracerKind.Console
  52.         }, EventShape.Point, StreamEventOrder.FullyOrdered);
  53.  
  54.     return query;
  55. }

Upon running this query with the supplied data sets, we observe:

 STREAM,Point,12:00:00.000        Activity = Logon        Location = Seattle        Name = Fred Jones        Status = Success        UserId = 1STREAM,Point,12:00:01.000        Activity = Logon        Location = Portland        Name = Bob Murphy        Status = Fail        UserId = 2STREAM,Point,12:00:05.000        Activity = Browse        Location = Seattle        Name = Fred Jones        Status = Success        UserId = 1STREAM,Point,12:00:06.000        Activity = Logon        Location = Portland        Name = Bob Murphy        Status = Success        UserId = 2

Comments

  • Anonymous
    October 19, 2010
    Helps a lot..Thanks Mark.
  • Anonymous
    June 01, 2011
    Hi,In the first note after Reference Data Stream (static data) you say: 'This is not a very real-world scenario – we’ll cover the additions required to incorporate changing data in the next blog post, showing how to pull data from SQL Server.'.Has this blog ever been posted? If so, can you provide a link to it, if not, when will it be posted?Regards,René