When processing data, it is common to perform data enrichment. Enrichment is useful when the data source contains only partial information, but the analytics require additional information that is available only in other data sources. The InfoSphere Streams database toolkit contains two enrichment operators: ODBCEnrich and SolidDBEnrich. These operators require the data used for enrichment to be in a database.

In this post, we illustrate how to develop an SPL composite that serves as a generic file-based enricher. In this solution, we use a FileSource to scan the enrichment data from a file, and then store it in an in-memory map in a Custom operator. This map is keyed by the attributes used to correlate incoming tuples with the enrichment data. If the enrichment data fully fits in memory, this solution can be more efficient than querying the database every time a tuple must be enriched.

The code below shows a sample invocation of the file-based enricher (operator FileEnrich). In this example, the program generates a stream called Data, which has attributes id and city. Data is then consumed by FileEnrich, which outputs an enriched stream using the id attribute as a key. The output stream contains both Data attributes (id and city) and EnrichT attributes (id and name). Note that because EnrichT and Data share the id attribute, id appears only once in the EnrichedData stream. The FileEnrich operator receives the following parameters:

• key - attribute in the input stream that is used for correlation
• enrichmentType - tuple type describing the enrichment data
• enrichmentKey - attribute of enrichmentType that is used for correlation. This parameter is passed as a string.
• enrichmentKeyType - type of enrichmentKey. This type must match the key type.
• enrichmentFile - file name containing the enrichment data.

namespace sample;

composite Main {
  type EnrichT = tuple<int32 id, rstring name>;
  
  graph
    stream<int32 id, rstring city> Data = FileSource() {
      param
        file: "in.dat";
    }
    
    stream<Data, EnrichT> EnrichedData = FileEnrich(Data) {
      param
        key: id;
        enrichmentType: EnrichT;
        enrichmentKey: "id";
        enrichmentKeyType: int32;
        enrichmentFile: "names.dat";
    }
}

We now show the code for the generic FileEnrich composite operator. This composite is developed using 2 primitive operators and 1 Custom operator. The first operator is a FileSource (line 11). The FileSource uses the enrichmentFile parameter and produces a stream of type enrichmentType. Using a parameter to establish the type of the FileSource output stream gives users the option to use a CSV file with any set of attributes. The second operator is a Switch (line 16), which serves exclusively to control when the input stream (In) can start flowing into the downstream operator. By default, this operator has an initial status of false (i.e., blocking tuples). The status parameter indicates the action taken once a tuple arrives in the second input stream. In this case, a true value indicates that the switch will open when a tuple arrives. The third operator (lines 21-22), implemented as a Custom, is the one responsible for doing the data enrichment itself.

namespace sample;

composite FileEnrich(input In; output Out) {
  param
    attribute $key;
    type $enrichmentType;
    expression $enrichmentKey;
    type $enrichmentKeyType;
    expression $enrichmentFile;
  graph
    stream<$enrichmentType> EnrichmentData = FileSource() {
      param
        file: $enrichmentFile;
    }
    
    stream<In> Data = Switch(In; Control) {
      param
        status: true;
    }
    
    (stream<In, $enrichmentType> Out;
     stream<int32 dummy> Control) = Custom(Data; EnrichmentData) {
      logic
        state: {
          mutable map<$enrichmentKeyType, $enrichmentType> dataMap;
          mutable $enrichmentKeyType mapKey;
        }
        onTuple EnrichmentData: {        
          mutable int32 ret = getTupleAttributeValue(EnrichmentData, $enrichmentKey, mapKey);
          if (ret == -1) {
            log(Sys.error, "Error getting tuple attribute value");
            shutdownPE();
          }
          insertM(dataMap, mapKey, EnrichmentData);
        } 
        onPunct EnrichmentData: {
          if (currentPunct() == Sys.FinalMarker)
            submit({dummy = 1}, Control);
        }
        onTuple Data: {
          mutable $enrichmentType enr = {};
          if ($key in dataMap)
            enr = dataMap[$key];
          mutable tuple<Out> outTuple = {};
          assignFrom(outTuple, Data);
          assignFrom(outTuple, enr);
          submit(outTuple, Out);
        }
    }
}

The Custom operator has two phases of execution. First, it builds a map based on EnrichmentData, the stream generated by the FileSource. To create this map, this operator uses the enrichmentKeyType and the enrichmentType itself (line 25). To populate the map, the operator uses the function getTupleAttributeValue (line 29) to get the value of a tuple attribute when that attribute is specified as a string. This function is defined as a native function so that we can use the C++ reflection APIs to inspect the specific tuple attribute and get its corresponding value. The XML segment below shows the SPL signature for the native function.

<function>
  <prototype>
    <![CDATA[<tuple T1, any T2> int32 getTupleAttributeValue(T1 tup, rstring attr, mutable T2 value)]]>
  </prototype>
</function>

The C++ code below shows the implementation of this function. This function returns an error flag in two cases: (i) if the attribute provided as a string does not exist, and (ii) if the attribute type does not match the type of value.

int32_t getTupleAttributeValue(SPL::Tuple const & tup, SPL::rstring const & attr, SPL::ValueHandle value)
{
  if (tup.findAttribute(attr) == tup.getEndIterator()) {
    SPLAPPTRC(L_ERROR, "Attribute " <<attr <<  " does not exist", "enrich");
    return -1;
  }
  SPL::ConstValueHandle src = tup.getAttributeValue(attr);
  if (!value.assignFrom(src)) {
    SPLAPPTRC(L_ERROR, "Attribute type and value type do not match: ", "enrich");
    return -1;
  }
  return 0;
}

The second stage of execution happens after the stream produced by the FileSource is fully processed, which is indicated by a final punctuation. At this point, the Custom operator notifies the Switch (line 38), and the data enrichment process starts. For enrichment, the Custom operator merges the attributes of Data and EnrichmentData using the assignFrom function (lines 45-46). This function assigns all matching fields from one tuple to the other, so be careful when naming the attributes of the enrichment type. If attribute names overlap, the last assigned value will prevail, which in this case is the value available in the enrichment tuple (line 46).

In summary, the FileEnrich composite has three characteristics that make it generic:

1. A type parameter (enrichmentType), which allows the enrichment file to have any stream type;
2. An attribute parameter (key) that allows users to specify externally which attribute of the input stream should be used as an enrichment key;
3. A string parameter, which allows users to choose which attribute of the enrichment type that must be used for enrichment correlation. Passing the parameter as a string allows users to externally reference a stream attribute name from a stream that is created only inside the composite. In addition, such string can be used by native functions, where one can leverage the power of the reflection APIs.

Thanks to Bugra Gedik for this example!

Identificações:  composite enricher stream_computing streams spl file operator