Streaming Split Join Pattern

About This Task

Your objective is to perform three enrichments:

• Retrieve a driver's certification and wage plan from the driver's table.

• Retrieve the driver's hours and miles logged from the timesheet table.

• Query weather information for a specific time and location.

To do this, use the split join pattern to split the stream into three, perform the enrichment in parallel, and then re-join the three streams.

Steps for Creating a Split Join Key

1. Create a new split key in the stream which allows you to join in a common field when you join the three stream.

   To do this, drag the projection processor to the canvas and create a connection from the EventType rule processor to this projection processor.

   When configuring the connection, select the Non Violation Events Rule which tells SAM to only send non-violation events to this project processor.

   

2. Configure the projection processor to create the split join key called splitJoinValue using the custom UDF you uploaded earlier called "TIMESTAMP_LONG".

   You will also do a transformation which calculates the week based on the event time which is required for one of the enrichments downstream. Configure the processor with the following parameters:

   

Steps for Splitting the Stream into Three to Perform Enrichments in Parallel

1. With the split join key created, you can split the stream into three to perform the enrichments in parallel.

   To do the first split to enrichment the wage and certification status of driver, drag the "ENRICH-PHOENIX" processor to the canvas and connect it from the Split project processor.

2. Configure the enrich processor with the following parameters:
   1. ENRICHMENT SQL: select certified, wage_plan from drivers where driverid=${driverId}
   2. ENRICHMENT OUTPUT FIELDS: driverCertification, driverWagePlan
   3. SECURE CLUSTER: false
   4. OUTPUT FIELDS: Click Select All.
   5. NEW OUTPUT FIELDS: Add new output fields for the two enriched values: driverCertification and driverWagePlan.

   After this processor executes, the output schema will have two fields populated called driverCertification and driverWagePlan.

   

3. Create the second stream to enrich the drivers hours and miles logged in last week by dragging another "ENRICH-PHOENIX" processor to the canvas and connecting it from the Split projection processor.

   

4. Configure the enrich processor with the following parameters:
   1. ENRICHEMNT SQL: select hours_logged, miles_logged from timesheet where driverid= ${driverId} and week=${week}
   2. ENRICHMENT OUTPUT FIELDS: driverFatigueByHours, driverFatigueByMiles
   3. SECURE CLUSTER: false
   4. OUTPUT FIELDS: Select the splitJoinValue field.
   5. NEW OUTPUT FIELDS: Add new output fields for the two enriched values driverFatigueByHours and driverFatigueByMiles.

      

   After this processor executes, the output schema will have two fields populated called driverFatigueByHours and driverFatigueByMiles.

5. Create the third stream to do weather enrichment by dragging the custom processor you uploaded called "ENRICH-WEATHER" processor to the canvas and connect it from the Split project processor.

   

6. Configure the weather process with the following parameters (currently the weather processor is just a stub that generates random normalized weather info).
   1. WEATHER WEB SERVICE URL: http://weather.com/api?lat=${latitude}&lng=${longitude}
   2. INPUT SCHEMA MAPPINGS: Leave defaults
   3. OUTPUT FIELDS: Select the splitJoinValue and the three model enriched features

      

   After this processor executes, the output schema will have three fields populated called Model_Feature_FoggyWeather, Model_Feature_RainyWeather, and Model_Feature_WindyWeather.

Steps for Rejoining the Three Enriched Streams

1. Now that you have done the enrichment in parallel by splitting the stream into three, you can now join the three streams by dragging the join processor to the canvas and connecting the join from the three streams.

   

2. Configure the join processor like the following where you use the joinSplitValue to join all three streams.

   For the Output field, click SELECT ALL to select all the fields across the three streams.

   

3. Now that you have joined three enriched streams, normalize the data into the format that the model expects by dragging the "NORMALIZE-MODEL-FEATURES" custom processor that you added to the canvas.

   For the output fields, select all the fields and leave the mapping as defaults.

   

Result

Your flow looks similar to the following.