Processors

Processors are built-in operators for processing the data by performing various transformations and analytical operations. Each processor is responsible for per­forming a specific action. You can drag and drop multiple processors and stitch them together to implement the business logic in a pipeline. 

Alert

Alert processor notifies users about occurrence of a specific event during pipeline execution as configured by the user.

Configuring Alert processor for Spark pipelines

To add an Alert processor into your pipeline, drag the processor to the canvas and right click on it to configure.D

Click on the NEXT button. Specify the Alert criteria here.

.

Click on the NEXT button. Enter the notes in the space provided. Click Save for saving the configuration details.

How to View Generated Alerts

To view generated Alerts, go to your Workspace > Alerts > Information.

 

The bell icon on top right of the page shows notification of alerts generated. For example, if you see 1 next to bell icon, it indicates 1 alert is generated.

If you click on the bell icon, you can view the alerts generated. The system will dis­play last twenty alerts at a time.

Refresh Interval: The time interval after which you want to refresh or reload the page. For example, if refresh interval is 5 minutes, page will be refreshed after 5 minutes.

Actions:

The following two types of actions can be performed on the Alerts.

1. View Description: The Alert description which you entered at the time of creation of Alert can be viewed by clicking on the eye  icon.

2. View Message Source: The source record for which alert is generated can be viewed by clicking on the View Message Source    icon.

 

Aggregation

The aggregation processor is used for performing the operations like min, max, aver­age, sum and count over the streaming data.

 

For example, you have an IOT device, which generates an event comprising of three fields: id, time and action.

 

You wish to calculate how many actions of each type happened in a window of 15 seconds duration. You can make use of Aggregation (COUNT) operation for calculat­ing the number of actions of each type from total events received in a window of 15 seconds duration.

The following table lists the functions supported by Aggregation processor.

Configuring an Aggregation processor for Spark pipelines:

To add an Aggregation processor into your pipeline, drag the processor to the can­vas and right click on it to configure.

 

Click on the Next button.You will view new schema based on your aggregation oper­ation applied.

 

Click on the NEXT button. Enter the notes in the space provided.

 

Click SAVE for saving the configuration details.

 

 

Custom Processor

It is a processor which enables you to create a custom business logic. It is used for components which are not inbuilt within StreamAnalytix.

Implements the ‘com.streamanalytix.framework.api.processor.JSONProcessor ‘inter­face and provides custom business logic in implemented methods.

You are able to perform operations on datasets which are not supported by other processors.

Pre-requisites

To use a custom processor, you need to create a jar file containing your custom code. Then upload the jar file in a pipeline or as a registered component.

To write a custom logic for your custom channel, download the sample project.

 

Import the downloaded sample project as a maven project in Eclipse. Ensure that Apache Maven is installed on your machine and that the PATH for the same is set.

Implement your custom code. Build the project to create a jar file containing your code, using the following command:

mvn clean install –DskipTests.

 

If the maven build is successful, you need to upload the jar on the pipeline canvas as shown below.

 

Custom processor for Spark

Custom processor for Spark supports writing code in Java programming language.

While using Custom processor in pipeline, you need to create a class which imple­ments com.streamanalytix.framework.api.spark.processor.CustomProcessor inter­face. Add unimplemented methods in this class to write your own business logic in it.

Shown below is a sample class structure:

 

There are three methods to implement.

1. Init: Enables to enter any initialization calls.

2. Process: Contains actual business logic.This method is called for each and every tuple.

3. Cleanup: All resource cleanup occurs in this method.

Configuring Custom processor for Spark pipelines

To add a Custom processor into your Spark pipeline, drag the Custom processor on the canvas. Right click on it to configure as explained below.

 

Click on the NEXT button. Enter the notes in the space provided.

 

Click SAVE for saving the configuration details.

Container

The container processor loads data to Couchbase server and reads data from it.

This processor is used as a caching container which reads both the aggregated data and raw data from the bucket.

Configuring Container processor for Spark pipelines

To add a Container processor into your pipeline, drag the processor to the canvas and right click on it to configure.

 

Click on the NEXT button. Enter the notes in the space provided.

Click SAVE for saving the configuration details.

Distinct

Distinct is a core operation of Apache Spark over streaming data. The Distinct pro­cessor is used for eliminating duplicate records of any dataset.

Configuring Distinct processor for Spark pipelines

To add a Distinct processor into your pipeline, drag the processor on to the canvas and right click on it to configure.

 

Enter the fields on which distinct operation is to be performed.

Click on the NEXT button. Enter the notes in the space provided.

Click SAVE for saving the configuration details.

Note: Distinct can't be used right after Aggregation and Pivot processor.

Example to demonstrate how distinct works.

If you apply Distinct on any two fields: Name and Age, then the output for the given fields will be as shown below: I

Dedup

In applications, you are often encountered with large datasets where duplicate records are available. To make the data consistent and accurate, you need to get rid of duplicate records keeping only one of them in the records.

Dedup processor returns a new dataset after removing all duplicate records.

Configuring Dedup processor for Spark pipelines

To add Dedup processor into your pipeline, drag the processor to the canvas and right click on it to configure.

Click on the NEXT button. Enter the notes in the space provided.

Click Save for saving the configuration details.

Example to demonstrate how Dedup works: 

You have a dataset with following rows:

[Row(name='Alice', age=5, height=80),
Row(name='Alice', age=5, height=80),
Row(name='Alice', age=10, height=80)]

Now if Dedup columns are [age, height], then Dedup processor would return below dataset:

[Row(name='Alice', age=5, height=80),
Row(name='Alice', age=10, height=80)]

and if Dedup columns are [name, height], then Dedup processor would return below dataset:

[Row(name='Alice', age=5, height=80)]

DQM

DQM stands for Data Quality Management. DQM processor defines the quality rules for data and performs necessary action on the faulty data.

It performs series of validation on individual data field along with actions on func­tions, in case if applied validation fails. Varieties of validations are available for each Data Type such as Not Null, Contains, and Starts With and Ends With.

DQM processor has six categories of Validation functions, one for each Data Type.

Data Type

1.String

 

2.Number

 

3.Decimal

 

4.Date

 

5.Boolean

 

6.Timestamp

 

DQM processor has three Action functions that are triggered when Validation fails:

1.Send To Error –It persists the invalid record in configured error handler target and discards the record for further processing in pipeline. Error handler becomes mandatory once you select “Send To Error” action. The key “errorMessage” in the error handler target contains the expressions due to which validation failed.

2.Assign Value –It gives option to assign static or dynamic values to any message field.

a.Static Values:

Select any message field and apply validation on it. If the validation criteria is  not met , assign static values to that field.  

Example :

Let us apply validation on field “id” (id equals to 10).

 

Click on the Save button for saving the validation criteria.

Whenever the value of the field “id” is equal to “10”, the record will be processed as it is  .

If the  value of the field “id” is not equal to “10” , static value 100 will be assigned to that id .

 

 

Click on the Save button for saving the configuration settings.

Click Next and write the notes. The changes will be reflected in Auto Inspect window

 

We can see in the above screen wherever id value was not 10, the value got replaced with the static value 100.

b.Dynamic values :

• Dynamic String Values:

• Dynamic Numeric values

Dynamic String Values

Dynamic values can be assigned using @ (the rate operator).

Select any message field and apply validation on it. If the validation criteria is  not met, assign dynamic values to that field.  

Example:

Let us apply validation on field “id” (id equals to 10).

 

 

Wherever the value of the field id is equal to 10, the record will be processed as it is  . If this validation criteria is not met , dynamic value  should be assigned to that id .

You can dynamically assign the values of any other field like crop, soil, climate, pest, cost_per_hectare (message fields) to the field id wherever validation criteria is not met.

Write @ (the rate operator) in the values text box   of Configuration Settings-Assign Value .

All the message fields will be displayed in the drop down list. Select the field whose value is to be dynamically assigned to field id .

 

For example, let us select field  “soil” from the drop down list.

Write the expression as  : @{soil}

The selected field value has to be enclosed within the curly braces.

After writing the expression, click on the Save button.

 

Wherever the value of  the field id is 10,  the record will be processed as it is.

Otherwise , the value of the field soil will be assigned to the id value in that column.

Click on the Save button for saving the configuration settings.

Click Next and write the notes. The changes will be reflected in Auto Inspect window

We can see in the below screen, wherever field id value was not 10, field soil got assigned to the id value.

 

 

Note: It is important to keep data type into consideration while assigning the dynamic values. The data type of two columns should be same for assigning the values except the String datatype.

Dynamic Numeric value: For numeric fields, mathematical operations can be used for assigning the values.

Example :

Let us apply validation on field “id” (id equals to 10).

Wherever the value of the field id is equal to 10, the record will be processed as it is . If this validation criteria is not met, assign dynamic value to the field id using arithme­tic operators.

 

Let us apply simple mathematical operation on the field “cost_per_hect”.

Write the expression as @{cost_per_hect+100}

The expression will be evaluated as cost per hectare  value will be added to 100.  The  output will be assigned to the field id wherever its value is not 10.

 

Click on the Save button for saving the configuration settings.

Click Next and write the notes. The changes will be reflected in Auto Inspect window.

 

Let us evaluate the columns “ id and cost_per_hect”.

The value of the f cost_per_hect  in the second row is 7193 , if 100  is added to it , the output is 7293  and this value gets assigned to the  field id .

The value of the f cost_per_hect  in the third  row is 7403, if 100 is added to it,the out­put is 7503, this value gets assigned to the field id .

Accordingly, all   id values get updated except where the id is 10.

3.Discard –It discards the record for further processing in pipeline.

If a record is eligible for more than one action then Actions will be executed as per below order:

1.Send To Error (Highest priority)

2.Discard (Lowest priority)

Configuring DQM Processor:

To add a DQM processor into your pipeline, drag the DQM processor to the canvas and connect it to a channel or processor. Right click on it to configure.

On the configuration canvas of Data Quality processor, select fields, then drag and drop the Validation and Action functions.

Choose a field in message panel from Configuration, respective data type validations will be expanded in function panel as shown in below screenshot.

 

After selecting the message, drag and drop the validation functions and connect them.

Connecting lines show the Logical operator, AND/OR over it.

By default, the AND operator is selected, but it can be toggled to OR operator via right click on it as shown below:

 

Most validation functions in Data Quality processor does not require user input such as Not Null, Is Empty, Upper Case etc, hence no error icon gets visible on top of them. There are some validation functions which requires user input and shows the error (red) icon for configuration.

Right click on the field (message) to configure the required input for validation func­tion.

Configuration panel also shows the Negate option so that same validation function can be utilized with reverse condition as shown below in the screenshot.

 

Once all of the validations are configured, you can add an Action function at the end. This action function gets triggered if validation fails as shown below:

 

You can select multiple fields and define the validations as defined for city field in above screenshot.

Note: If you are using any Timestamp field for configuration, specify the value in fol­lowing format:

“yyyy-MM-dd HH:mm:ss”

For Date field, use below format:

“yyyy-MM-dd ”

Drop

Drop Processor removes the selected fields from the Data Sets, which means it will not be reflected in the Schema.

Configuring Drop processor for Spark pipelines

To add a Drop processor into your pipeline, drag the processor to the canvas and right click on it to configure.

 

Click NEXT and Edit Schema if required.

 

Click NEXT and add notes if required.

Click SAVE for saving the configuration details.

Expression Evaluator

This processor is responsible for performing transformation operations on incoming dataset, e.g., replace, format, trim, uppercase, lowercase, etc. It uses spark expres­sion language for preparing transformation query.

Configuring Expression Evaluator for Spark pipelines

To add an Expression Evaluator processor into your pipeline, drag the processor to the canvas and right click on it to configure.

All the columns of the schema are populated and you can apply any transformation on it. The functions that can be applied are all listed in the right table, as shown in the figure.

 

Expression Filter

This processor is responsible for performing filtering operations on incoming dataset, e.g., equals, contains, not-null, ranges, matches, starts-with, ends-with, etc. It uses spark expression language for constructing filter criteria.

Configuring Expression Filter for Spark pipelines

To add an Expression Filter processor into your pipeline, drag the processor to the canvas and right click on it to configure

 

.

You can apply any transformation and the list of functions that can be applied are all listed in the right table, as shown in the figure.

Filter

Data filtering is required to refine the datasets so that you simply get what you need excluding other data that can be repetitive, irrelevant or sensitive. Different type of filters can be used to query results or other kind of information.

The Filter processor is used to display specific records by using filter criterion. For example, you can define the Filter rule: grade contains “software” and name contains “Mike”. The system will fetch all the records which meet this criteria.

Configuring Filter processor for Spark pipelines

To add a Filter processor into your pipeline, drag the processor to the canvas and right click on it to configure.

Click on the NEXT button.

 

Click on the Save button If you do not wish to create a new template.

Click on Save as Template button for creating a new template by using existing com­ponent configuration.

 

A window pops up with the name Filter-Save as window.

Specify the Template name and Scope.You can select the scope as Global or Work­space.

Click on the SAVE button after entering all the details.

A template will be created with the given name and scope.

It will be visible under Load from Template drop down list and on Component Man­agement page.

Note: Template name should be unique.

If a template has a workspace scope, same template name cannot be used within a workspace but same name can be given in any other workspace.

If a template has a global scope, same template name cannot be used in any other workspaces.

 

Functions

Function processor enables to perform spark SQL functions on dataset.

Functions processor supports:

• MVEL expressions

• lookup functions

• Date

• String

• Math

• Miscellaneous functions

Configuring Functions processor for Spark pipelines

To add Functions processor into your pipeline, drag the processor to the canvas and right click on it to configure as explained below:

 

Click on the NEXT button. You will view the fields coming from source or enriched via functions processor .

 

You can provide new output format of the  date field . For example, if the input date format was yyyy-dd-yy , specify new date format as yyyy/MM/dd

Click Next after changing the Date format . Enter the notes in the space provided.

 

Click SAVE for saving the configuration details.

Please refer Functions guide for detailed information.

Join

A Join processor combines columns of two incoming messages (DataSets). It creates a set that can be saved as a table or used as it is. A Join is used for combining col­umns from two tables by using values common to each table.

Configuring Join processor for Spark pipelines

To add a Join Processor into your pipeline, drag the processor on the canvas and right click on it to configure.

First tab is Join Condition. It allows to define the Join Type between two DataSets where two columns from both DataSets can have a Join condition applied to them.

The available Join Types are:

Second Tab is Filter Condition, It allows to apply a filter condition to the DataSets. You can apply AND and OR operations with two filter conditions.

 

The Filter Condition that can be applied are as follows:

l contains

l not contains

l begins_with

l ends_with

l equal

l not_begins_with

l not_ends_with

l is_null

l is_not_null

l in

l not_in

l matches

l custom

l not_equal

In case of any operator, you can provide a custom expression as a right-side oper­and, instead of selecting from the drop down.

In case of Left Anti and Left Semi (Join Type), columns of only left side table are enlisted, for both the operand.

Third tab is Join Projection where you can apply the SELECT Clause along with cus­tom expressions and add a column to the columns.

Comma seperated columns and custom expressions can be availed as input data for the next component.

 

Jolt

Jolt is a JSON transformation library written in Java where the specification for the transformation is a JSON document.

Jolt processor is used for performing a JSON to JSON transformations in Java. It applies transformation on Input JSON document and generates a new JSON docu­ment as per the Jolt specificationsl

Jolt processor transforms the structure of JSON data and does not manipulate the values.

Each transform has it's own Domain Specific Language which facilitates it’s job.

Refer following links for detailed information:

https://github.com/bazaarvoice/jolt

https://jolt-demo.appspot.com/#inception

Configuring Jolt processor for Spark pipelines

To add a JSON Processor into your pipeline, drag the processor on the canvas and right click on it to configure.

 

Select the Jolt Specification JSON file which is to be uploaded. Click on UPLOAD FILE button. Browse to the location where JSON file is saved and double-click on it for uploading the JSON file.

The content of the Specification JSON file gets displayed below. This specification gets applied on incoming json document and generates the transformed output.

Additional properties can be added using ADD CONFIGURATION link.

Click on the NEXT button. Enter the notes in the space provided. Click Save for saving the configuration details.

Limit

The Limit processor is used to fetch specified number of rows from the dataset.

For example, if 1000 records are getting generated and you want to fetch only 50 rows, you can make use of this processor.

The record of these 5o rows will be carried to the next component.

This processor can only be used with Batch components.

Note: If you try to add Limit processor after Streaming channel, it will not get added and you will get a warning message.

Configuring Limit Processor for Spark Pipelines

To add Limit processor into your pipeline, drag the processor to the canvas and right click on it to configure as explained below

 

Specify number of rows which are to be fetched in “No of Elements “ fields.

Additional configuration details can be added using ADD CONFIGURATION link.

Click on the Next button. Enter the notes in the space provided. Click Save for saving the configuration details.

NA

The NA processor is used to handle the null values of any dataset.

If the value specified is Number, all null values of selected input field will be replaced by the specified number.

If the value specified is String, all null values of selected input field will be replaced by the given string.

For example, if there is any dataset in which input fields “Name”  and “Age” are null, you can specify value for Name as “Mike”  and “Age” as “30” that should substitute the null values.

Note: It does not support Date and Timestamp fields.

Configuring NA processor for Spark pipelines

To add NA Processor into your pipeline, drag the processor to the canvas and right click on it to configure.

 

Click on the NEXT button. Enter the notes in the space provided.

Click SAVE for saving the configuration details.

 

Pivot

Pivot is an aggregation processor where one of the grouping columns have distinct values which later gets transposed to form individual columns. These column values further help in data analysis and reporting.

Configuring Pivot processor for Spark pipelines:

To add Pivot processor into your pipeline, drag the processor to the canvas and right click on it to configure.

Click on the NEXT button.You will view new schema based on your pivot configuration on Schema tab.

 

Click SAVE for getting the final output of pivot processor.

 

Example:

Let us say, you have a dataset comprising of five records having four fields: “OrderId”, “Product”, “Category” and “Amount”.

 

You wish to group on the basis of field “Product”, pivot on column “Category” and type of aggregation to be applied is “SUM” on column “Amount”.

The Pivot processor applies the aggregation operation(SUM) on the Pivot column and creates pivot table as shown below:

   

Since, you have applied the operation “SUM” on column “Amount”, it will calculate the sum of different vegetables and fruits.

Sum of the fruits banana is 300, vegetable broccoli is 400 whereas carrot is 100.

If no amount is found for a particular vegetable, it will show null values against the product.

Python

The Python processor allows you to perform following operations:

• Write custom Python Spark code for defining transformations on Spark Data­Frames.

• Write custom Python code for processing input records at runtime.

Configuring Python processor for Spark pipelines

To add a Python processor into your pipeline, drag the processor on the canvas and right click on it to configure.

Let us see how to pass configuration parameters in Python processor.

You can provide configuration parameters in Python processor in form of key value pair. These parameters will be available in form of dictionary in function given in Function Name field as second argument. So function given in field Function Name will take two arguments: (df, config_map)

Where first argument will be dataframe and second argument will be a dictionary that contains configuration parameters as key value pair.

In below example HDFS path of trained model is provided in configuration parameter with key as ‘model_path’ and same is used in predictionFunction.

 

If there is any error/exception in python code, then you will get an error/exception message on user interface.

Example:

A simple program is written in text area in Python processor. Here the variable model is used but not defined in program, hence an exception with message ‘global name model is not defined’ is displayed on the screen.

 

Click on the NEXT button after specifying the values for all the fields.

Enter the notes in the space provided.

Click on the SAVE button after entering all the details.

Repartition

Repartition processor divides large datasets into multiple parts.

The Repartition processor is used when you want to increase or decrease the paral­lelism in an executor. The number of parallelism maps to number of tasks running in an executor. It creates either more or fewer partitions in order to balance data across them and shuffles data over the network.

For example, you can use single partition if you wish to write data in a single file.Mul­tiple partitions can be used for writing data in multiple files

Configuring Repartition processor for Spark pipelines

To add a Repartition processor into your pipeline, drag the processor on the pipeline canvas and right click on it configure.

 

Enter the number of partitions in the Parallelism field.

Click on the NEXT button. Enter the notes in the space provided.

Click on the SAVE button after entering all the details.

Register as Table

This processor is used for fetching historical data from any streaming or batch source and that data gets registered in the table. This table can be referred further if you wish to perform some queries on registered data sources.

For example, let us say, you have any historical data on HDFS which contains infor­mation about different departments of an organization. You can register that data using Register as Table processor and that registered table can be used in SQL pro­cessor to fetch number of employees in an organization.

Configuring Register as Table processor for Spark pipelines

To add a Register as Table processor into your pipeline, drag the processor on the canvas and right click on it to configure.

 

If data source selected is HDFS, you will view following fields:

If you select the Data Source as HIVE or JDBC, you will view two additional fields:

• Database Table Name

• Execute Query

If Data Source selected is S3, you will view one additional field:

• Bucket Name:

If the option selected is incoming message, output of sources connected before this processor will act as an incoming message for Register As Table processor.

You need to specify the name with which table is to be registered after fetching the data.

Rename

Rename processor enables renaming the Fields. The renamed columns will be reflected in the Schema.

Configuring Rename processor for Spark pipelines

To add a Rename Processor into your pipeline, drag the processor to the canvas and right click on it to configure as explained below.

 

Click NEXT to go to SCHEMA page. The renamed fields will be reflected in the Schema Tab.

 

Click on NEXT button. Enter the notes in the space provided. Click on the SAVE button for saving the configuration details.

Select

The Select processor is used to retrieve specific columns from the dataset.

 

On configuration settings page, specify the schema fields that you wish to retrieve in the output.

For example, you entered the fields: OrderId, ShipVia and Freight.

Only these three fields will flow to the next component which is connected in the pipeline.

Click on NEXT button. Enter the notes in the space provided. Click on the SAVE button for saving the configuration details.

 

SQL

SQL Processor allows you to run SQL queries over streaming data and registered tables.

It provides a common way for accessing different data sources.

For example, if you want to analyze weather data stream continuously and find the average temperature which is recorded every minute, you can use the SQL query. The output would be stream of new records showing average temperature record­ings.

Configuring SQL processor for Spark pipelines

To add a SQL processor into your pipeline, drag the processor to the canvas and right click on it to configure as explained below.

Click on the NEXT button.Enter the notes in the space provided.

 

Click SAVE for saving the configuration details.

Scala

Scala is a new language and is based on Java Virtual machine compatibility.You can code Scala in the similar manner as you code Java and has minimal prerequisites. Writing code in Scala is much faster and easy to use.

The Scala processor is used for writing custom code in Scala language.

Configuring Scala processor for Spark pipelines

To add a Scala processor into your pipeline, drag the processor on the canvas and right click on it to configure.

Next, click on the Jar Upload tab. Jar file is generated when you build Scala code. You can upload the third party jars from here so that the API's can be utilized in Scala code by adding the import statement.

Click on the NEXT button. Enter the notes in the space provided.

 

Click SAVE for saving the configuration details.

Sort

Sorting is arranging the data in ascending or descending order.

For example, you can use this processor for sorting the height of students in ascend­ing order.

Note: Currently, sorting is supported for ascending order only.

Configuring Sort processor for Spark pipelines

To add a Sort processor into your pipeline, drag the processor to the canvas and right click on it to configure.

Click on the NEXT button. Enter the notes in the space provided.

Click SAVE for saving the configuration details.

Note: In Batch  pipelines, Sort processor can be used without Aggregation proces­sor.Whereas in streaming pipelines, Sort processor can be used only after Aggrega­tion processor.

Union

Union processor is not configurable.

The Union operation is used when you want to combine elements of two datasets.

Example to demonstrate how union works.