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.

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 sav­ing 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. 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 provide custom business logic in implemented methods.

You will be 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 Data Source, 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. An option to Save as Template will be available. Add notes in the space provided and click on Save as Template.

Choose the scope of the Template and Click SAVE for saving the Template and con­figuration details.

Decoder

Fields are decoded to hide sensitive data.

Base64 is a group of binary-to-text encoding schemes that represent binary data in an ASCII string format by translating it into a radix-64 representation.

Base64 encoding schemes are commonly used when there is a need to encode binary data that needs be stored and transferred over media that are designed to deal with textual data. This is to ensure that the data remains intact without modifica­tion during transport

Configuring Decoder for Spark pipelines

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

Click Next to get the schema detected. Click next to add Notes and then save the configuration.

Decryption

Decryption processor is used to decrypt data coming from data source.

Configuring Decryption processor for Spark pipelines

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

Algorithm Types

After the decryption, the Detect Schema window will show the decrypted data.

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.

NOTE: This processor does not work on Mozilla Firefox.

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 seven categories of Validation functions, one for each Data Type.

Data Type

1.Array

Note: Array represents the JSON (not nested) array of primitive types (String, Num­ber, Double and Boolean)

2.String

3.Number

 

4.Decimal

 

5.Date

 

6.Boolean

 

7.Timestamp

 

8.Opening Braces and Closing Braces

Use these braces for adding or separating two or more expressions.

 

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

1.Send To Error –It persists the invalid records in configured error handler target and discard the records for further processing in pipeline. Right click the Send To Error button to open the error handler configuration.  When the data validation fails, a json message is sent to the configured error handler target. This json has the key “errorMessage” in the error handler target, that contains the expressions due to which validation fails.

Note: The last Error Log Target configuration saved on Send to Error is where the configuration is saved.

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 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 cost_per_hect in the third row is 7403, if 100 is added to tithe output 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 Data Source 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.

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.

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 func­tion 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”

Drools

Using Drools Processor we can perform rules based processing on incoming data using Drools Rule Engine.

To use this processor user needs to create a jar either manually or using Drools Workbench (Kie Jar) and upload it on the processor.

Configuring Drool processor for Spark pipelines

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

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 add notes if required.

Click SAVE for saving the configuration details.

Encoder

Fields are encoded to hide sensitive data.

Base64 is a group of binary-to-text encoding schemes that represent binary data in an ASCII string format by translating it into a radix-64 representation.

Base64 encoding schemes are commonly used when there is a need to encode binary data that needs be stored and transferred over media that are designed to deal with textual data. This is to ensure that the data remains intact without modifica­tion during transport.

Configuring Encoder for Spark pipelines

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

Click Next to get the schema detected.

Click Next to add Notes and save the configuration.

Encryption

Encryption processor allows data to be encrypted. The data coming out from a data source, which requires further processing and data security, this processor allows to hide the actual data during transformation or processing.

This is useful to hide or transform sensitive or confidential data such as credit card details, bank details, contact details, personal details etc. Encryption processor lets you to encrypt that sensitive information.

Configuring Encryption for Spark pipelines

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

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.

Note: Refer Expression Evaluator section in Data Preparation section.

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

.

Note: Refer Expression Filter section in Data Preparation section.

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 criteria. 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.

Field Splitter

The Field Splitter splits string data based on a regular expression and passes the separated data to new fields. Use the Field Splitter to split complex string values into logical components.

Field Splitter splits string data based on a regular expression and passes the sepa­rated data to new fields. Use the Field Splitter to split complex string values into log­ical components.

For example, if a field contains an error code and error message separated by a comma, you can use the comma to separate the code and message into different fields.

Configuring Field Splitter processor for Spark pipelines

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

The Field Splitters are as follows:

Field Converter

To edit the Data Type of the schema, you can use this processor. The Field Converter processor generates the schema and the fields are editable after the schema is detected.

Configuring Field Converter processor for Spark pipelines

To add Field Converter processor into your pipeline, drag the processor to the can­vas and right click on it to configure as explained below.

On the first page you will be asked to Add Configuration, which is optional.

Once you click Next, the schema is detected and you can edit the Data Type as required.

Field Replacer

A field replacer allows you to replace a field value by providing a hard coded value e.g. null with "some value", replace a value which is following a pattern by providing regex, and replace a field value by other field value or combination of field values.

You can replace input field if the regex pattern matches.

There are two option to provide values.

1. Value will be a combination (concatenation) of two columns.

For e.g: in below image it is “JOBROLE” CONCAT “NAME”.

Order in which you select the columns will be considered for the combination.

2. Enter a single value to replace with.

For e.g: “uservalue” is given

Configuring Field Replacer processor for Spark pipelines

To add a Field Replacer processor into your pipeline, drag the processor to the can­vas and right click on it to configure as explained below:

After configuring the component, detect the schema for desired result.

The below image confirms the input field (NAME) where regex pattern (sagar|Nikhil) matches replaced with uservalue and created new column (NEWNAME).

Field Flattener

Field Flattener allows you to flatten a nested record to produce a record with no nested fields, it can also flatten specific list or map fields.

By default, the new flattened fields is separated by”_”

Let’s take an example of a nested JSON.

Specify a path of input json in the record, so that it flattens all nested structures until the nested path is flat.

For example:

Flatten Example 1: The JSON below has nested fields under Employee, ID, Name, Permanent and Address. Now using the Field flattener, you will be able to flatten the nested fields i.e., Street, City, Zipcode as shown below:

{

  "Employee":{

     "Id":123,

     "Name":"Steve Rogers",

     "Permanent":true,

     "Address":{

        "Street":"El Camino Real",

        "City":"San Jose",

        "Zipcode":55014

     },

     "Role":"Developer"

  }

}

 

Configuring Field Flattener processor for Spark pipelines

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

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 generated from data source. 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.

HTTP

HTTP Processor is used to enrich the incoming data with the response of external rest calls. The user can consume different URLs in parallel and use their response to enrich the resulting data either in a new column or existing column.

 

Note:

The user can create and run multiple URLs by clicking on the ADD FIELD option.

Hashing

Hash fields based on different Hashing Algorithms, to hide sensitive information.

Hashing is a technique in which an algorithm (also called a hash function) is applied to a portion of data to create a unique digital “fingerprint” that is a fixed-size variable. If anyone changes the data by so much as one binary digit, the hash function will pro­duce a different output (called the hash value) and the recipient will know that the data has been changed. Hashing can ensure integrity and provide authentication as well.

The hash function cannot be “reverse-engineered”; that is, you can't use the hash value to discover the original data that was hashed. Thus, hashing algorithms are referred to as one-way hashes. A good hash function will not return the same result from two different inputs (called a collision); each result should be unique.

Configuring Hashing processor for Spark pipelines

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

After configuring all the fields, click Next; the schema will be detected and then you can verify and save the configuration.

Join

A join processor can create a relation between two incoming messages (DataSets).

Note: Join Processor has a feature called “Remove Stale Configuration”.

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.

Second Tab is Broadcast Table, it guides Spark to broadcast the specified table, when joining them with another table.

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 wit 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

Third tab is Join Projection where you can apply a query and expressions to the col­umns.

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 specifications

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 JOLT 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 Data Source, 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.

Masking

Field masking is done to hide sensitive data.

Configuring Masking processor for Spark pipelines

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

Configurable fields are as follows:

The screenshot shown below has two fields, one with Masking Type and other has Input Field to be masked.

 

Click next to get the schema detected.

As shown above, ALL the characters of field fieldname are masked and Head Cha­raters of newcol field.

Click Done to save the configuration.

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 can view the 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.

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.

StreamAnalytix provides support for two python versions, Python 2 and Python 3. Multiple version support enables a python processor to run on different python ver­sions.

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. To select a version from Python 2 and Python 3, select either from Python Version property.

NOTE: Code snippets contains sample codes for your reference.

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, an error/exception is generated.

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.

Rank

Use rank functions (rank, dense_rank, percent_rank, ntile, row_number) while config­uring the Rank processor.

Rank Processor computes the rank of a value in a group of values. The result is one plus the number of rows preceding or equal to the current row in the ordering of the partition.

dense_rank() - Computes the rank of a value in a group of values. The result is one plus the previously assigned rank value. Unlike the function rank, dense_rank will not produce gaps in the rankings.

percent_rank() - Computes the percentage ranking of a value in a group of values.

ntile(n) - Divides the rows for each window partition into n buckets ranging from 1 to at most n .

row_number() - Assigns a unique, sequential number to each row, starting with one, according to the ordering of rows within the window partition.

Configuring Rank processor for Spark pipelines

To add a Rank processor into your pipeline, drag the processor on the pipeline can­vas and right click on it configure.

Click on NEXT button. Enter the notes in the space provided. Click on the SAVE but­ton for saving the configuration 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 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 to 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 Done 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 as a table. This table can be referred further if you wish to perform some queries on registered data sources.

You can fetch tables from Cassandra, Couchbase, Redshift, HDFS, Hive, JDBC, Snow­flake, File, Incoming Message and S3.

For example, you have historical data on HDFS which contains information about dif­ferent departments of an organization. You can register that data using Register as Table processor and that registered table can be used in SQL processor 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, there will be following fields:

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

• Database Table Name

• Execute Query

If the user selects Snowflake, the below mentioned field will be additional:

If Data Source selected is S3, there will be 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.

If you select the Data Source as Cassandra, there will be two additional fields:

• KeySpace Name

• Cassandra Table Name

 

If you select the Data Source as Couchbase, there will be an additional field:

• Bucket Name

If you select the Data Source as Redshift, there will be a few additional fields, depending on the two following options:

l Database Table Name

l Execute Query

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 on NEXT button. Enter the notes in the space provided. Click on the SAVE but­ton 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.

Snowflake

Snowflake is a cloud-based data warehouse system. The user can use Snowflake as a processor in StreamAnalytix for ELT pipelines.

 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.

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.

 

Stored Procedure Processor

The Stored Procedure processor is used to run a set of SQL queries. The user can now invoke stored procedure and functions defined in the database within Strea­mAnalytix. While the input data is passed as an argument to functions and stored procedures, the resultant output can be used to enrich the incoming data.

Instead of sending multiple SQL statements to the database server, an application can send a set of queries in what is called a stored procedure.

A function is a set of SQL statements that perform some operation and return a result.

 

 

TopNRecords

Top-N queries are queries that limit the result to a specific number of rows. These are often queries for the most recent or the “best” entries of a result set. For efficient execution, the ranking must be done with a pipelined order by. You can get Top-N records based on a sorting criterion.

Top N Records works with Batch Pipeline only.

Configuring TopNRecords processor for Spark pipelines

To add a TopNRecords processor into your pipeline, drag the processor to the can­vas and right click on it to configure

Click Next to detect schema and Save the configuration.

 

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.