Data Pipelines

Pipeline Configuration

The home page of data pipeline is explained below:

With a StreamAnalytix Data Pipeline, you can perform the following actions:

• Design, upload and download real-time data pipeline.

• Drag, drop and connect operators to create applications.

• Create Datasets and reuse them in different pipelines.

• Create Workflows using pipelines and apply control nodes and actions on them.

• Create models from the analytics operators.

• Commit the Data Pipeline on GIT.

• Monitor detailed metrics of each task and instance.

• Run PMML-based scripts in real-time on every incoming message.

• Explore real-time data using interactive charts and dashboards.

Creating a Spark Pipeline

If you are creating a data pipeline for the first time, you need to ensure that you are in a new workspace. Click Manage Workspaces on the left navigation panel. You can create a new workspace or enter in an already existing workspace. Now, go to the Data Pipeline page from the left navigation panel. In this panel, you may create a new pipeline or edit/work in an existing pipeline.

To create a new pipeline, click on the +icon at the right of the screen. You are ready to create a pipeline on the grid-canvas.

Make sure that the Local/Livy connection is established. The green arrow above the pipeline canvas denotes that the connection is established.

Drag components on the pipeline builder (pipeline grid-canvas) from the right panel and connect the components.

Select and right click on a component to configure its properties (view Data Sources, Processors, Data Science and Emitters sections, to understand the components).

Once the configuration is done, click on the floppy icon to save the pipeline. (shown above)

Enter the details of the pipeline in Pipeline Definition window. (shown below).

The pipeline definition options are explained below:

 

 

When you click on More, the below mentioned configuration appears (they are same for Local or Livy).

Note:

• When you create a pipeline, the associated or used schema is saved as metadata under message configuration.

• Components are batch and streaming based.

• Likewise, in Spark standalone the application cores must be greater than or equal to the executor cores. as shown below:

Error Handler Configuration

Error handler configuration feature enables you to handle the pipeline errors.

Error log target are of three types: RabbitMQ, Kafka and Logger.

Error Handler helps you to log pipeline errors to logger file or on Kafka or RMQ. You can refer the log file or message queues to view those errors whenever required. All these errors will be visible on graphs under Summary view of pipelines under Appli­cation Errors Tab.

StreamAnalytix gives you an option to handle errors after your data pipeline is cre­ated. When you save a pipeline, specify whether you want to enable error handler or not. Either Kafka or RMQ is selected as an error log target. Whenever an error hap­pens, the pipeline will log the errors to the configured error handler. By default, errors will always be published to log files. If you want to disable logging errors to these queues, disable error handler while saving or updating actions on the pipeline. If error handler is disabled, error graphs will not be visible and errors will not be avail­able on the errors listing page.

NOTE: If RMQ or Kafka is down, the data pipeline will work. For the visibility of error graph default Elasticsearch connection should be in working state. Also, for DQM processor it is mandatory to keep this flag checked to use the Send to Error Feature.

Based on error handler target, you will be asked for target specific configurations like connection name, partitions and topic administration in case of Kafka and ttl in case of RabbitMQ.

Once you save or update the pipeline, and start the pipeline, you can view errors in the configured queue or topic as well as in pipeline logs.

NOTE: By default pipeline errors will be logged in Logger.

For viewing error graphs, go to the Data pipeline tab on Data Pipeline Home page, click on the three dots of the pipeline tile and click on View Summary.

Actions on Pipeline

To view the screen as shown below, go to the home page of Data pipeline and click on the three dots of a pipeline’s widget.

As shown in the snippet below, the following actions can be performed on a pipeline

.

Pipeline Run History

The user can monitor the batch pipeline activity and keep a track of the run details for each instance such as the total number of records in input and output process and time taken by the pipeline to run completely. After the successful run of the pipeline, click the ellipses of the pipeline tile to view the batch monitoring history.

In the pipeline history page, user will be able to view the batch monitoring table and the run history of the pipeline with that table.

 

 

 

Note: If there are multiple files that the user is reading then the entire list of file names along with locations will be displayed.

Pipeline Management

StreamAnalytix provides tools, such as Auto Inspect, Upload Pipeline, Data Prepara­tion, Download sample project, and Monitor, to manage pipelines at all the stages of its lifecycle.

Note: If a Inspect Session is not established using either LIVY or LOCAL; as soon as you start Auto Inspection, a Livy connection is established.

Livy

Livy connection can be created at any time and from anywhere in the workspace.

However, you can disconnect from it after the pipeline has been created. It is not mandatory for Livy to be in an initialized state for a pipeline to be edited and saved.

While connecting the Livy session, you will be prompted for Extra Java Driver Options, Extra Java Executor Options and Extra Spark Submit Options. All are non-mandatory fields.

There is a place holder property displayed in the text areas of Extra Java Driver Options and Extra Java Executor Options i.e. -Djava.io.tmpdir:/tmp. You can configure this property to change the target folder for Spark Client temporary files.

While saving the pipeline, these parameters can be populated by clicking on the “Populate from Livy session” link given on Pipeline Definition window.

Logs will take you to the Livy Inspect Session logs. These logs are not live logs.

These parameters are disabled for a superuser.

Local Inspect

Local Inspect connection can be created at any time and from anywhere in the work­space. However, if you do not want to create a connection explicitly, the connection would be initialized automatically when you start Auto-Inspection. However, you can disconnect from it after the pipeline has been created. It is not mandatory for Local Inspect to be in initialized state for a pipeline to be edited and saved.

While connecting the Local Inspect session, you can specify additional extra java options by clicking the Spark Parameters option. This parameter is optional.

In Spark Parameter, you can supply Extra Java Options for the inspect JVM process. These parameters are disabled for a superuser.

Auto Inspection

When your component is inspect ready, slide the Auto Inspect button.

Auto Inspection lets you verify the component’s data with the schema as per the configuration of the component. This happens as soon as the component is saved.

Auto inspect is done manually therefore while editing the pipeline, you can choose to not inspect the pipeline. When you edit a pipeline, auto inspect will be off by default.

While editing, you can add/remove new components in the pipeline without auto inspect.

You can auto inspect the properly configured components of the pipeline if the pipe­line is not entirely configured, as shown below:

Also, the auto inspection can be stopped at any point, even while schema is being detected, and pipeline can be saved even if the inspection hangs or inspect fails.

Clicking on this button stops the auto-inspection of an existing pipeline.

Note: Once the Auto Inspect is enabled, a Livy connection is initialized. (If it is not con­nected).

If one of the component is not configured, the auto inspect will run for the configured components only. The auto inspect window will show a notification in Red for the uncon­figured component.

A single component can be inspected by clicking on the eye icon on top of the com­ponents as shown below:

 

If a pipeline is saved without inspection, the pipeline tile on the data pipeline homep­age will show the warning-icon of not inspected, Once you clilck on the exclamation mark, it shows the Pipeline Notifications, as shown in figure below:

Once the data is loaded from the source it is automatically played through the pipe­line. During auto inspect, if the schema gets changed due to any configuration changes, it will not reset the next components however it will show an exclamation mark.

This is how the corresponding configuration, logic and results from each operator is validated while the pipeline is being designed.

Data Preparation in StreamAnalytix

Data preparation is the process of collecting, cleaning and organizing data for analy­sis. Prepare the data with operations and expressions to generate customized data. Graphs and statistics are also used to present the prepared data.

To perform data preparation, configure a Data Source (For example, Kafka). Once the Data Source is configured and saved, you can run inspection on the columns of the schema that are reflected in the Data Preparation window. This allows you to build the pipeline while you are interacting with the data.

To perform Data Preparation, click on the eye icon of any component (Data Source, Processor or Emitter). A sub- section on the pipeline canvas opens with the component’s schema displayed in Data Preparation window.

In the example below, an Agriculture data file is uploaded. It holds information about crop, soil, climate, pests, season, average_yield, cost, high_production_state, crop_wastage and fertilizer.

By default, the data is displayed in Summary view with Profile Pane on the screen. (as shown below).

Data Pane: The schema takes the form of columns and is divided in records (as shown below). You can apply operations on an entry of a column or any possible combination, corresponding to which a new expression is added, which keeps creating your pipeline as in when you apply more actions.

The data preparation window has the following options:

Data preparation window has type2 more views: Summary View and Detailed View

Summary View shows the summary of number of records out of total records and also the columns in each record. It also shows the name of the current schema.

The Detailed View shows details of a selected column value and its corresponding rows.

Detailed View has eight sections. All the sections are displayed and explained below:

An important and elaborate section amongst the ones mentioned in the table above is Operations.

Operations

This option is at the top right of the column. The gear icon, which opens a sliding win­dow, shows a list of operations (shown in image below) that can be performed on the respective column of the schema.

Whenever any operation is applied on a column (s), a step is created and reflected in the left section of Data Preparation Window. Also, a processor is added to the pipeline, depending on the operation applied.

These data preparation processors are defined below in data preparation processors section.

  a. Filter

Filter the values based on a set of criteria (shown in image below) such as, Equals, Contains, Regex match and so on. Custom filter can be used for a custom value.

 b. Transforms

Filter the values based on a set of criteria (shown in image below) such as, Equals, Contains, Regex match and so on. Custom filter can be used for a custom value.

  c. Missing Value Replacement:

Replace the missing or null values with either Literal or Expression value.

Literal: Replaces Null and/or empty string values with a specified string literal.

Expression: Replaces Null and/or empty string values with a specified expression value.   

   d. Analyze

Under Analyze, you can view the Null values of the selected column.

Under Statistics, you can view the mathematical statistic/value of the entire column in the form of:

l Minimum

l Maximum

l Mean

l Median

l Standard Deviation

l Mode

l Distinct

l Sum

l Range

e. Pivot

Pivot operation is a kind of aggregation where type1 (or more) of the grouping col­umns has its distinct values transposed into separate columns.

Original Data

Pivot Operation in StreamAnalytix

l Pivot column – C

l Grouping fields – A, B

l Pivot values – small, large

l Type of aggregation – Sum

l Aggregation Column - D

Pivoted Data:

g. Group By

You can apply aggregation on a column using Group By operation and save the out­put in a field with or without grouped columns.

 h. Rename Column

Rename the column name.

 i. Create New Column

Create a new column using this filter.

  j. Remove Column

Remove the selected column.

Data Preparation Processors

Data Preparation operations add following processors in your data pipeline:

• Expression Filter

• Expression Evaluator

• Rename

• Select

• Drop

Data Preparation steps appear on left side of inspect window, as a sliding panel.

The data preparation operations when performed keep getting added to the data pipeline and are reflected in data preparation steps.

Even when data preparation steps are added manually in a pipeline, they are reflected in data preparation steps.

Expression Filter:

Expression Filter enables filter operations on the data column.

This processor gets automatically added on the canvas once you apply the filter operation.

For example, let us add the filter criteria as “crop” equals “cereals”.

 

Expression Evaluator

Expression Evaluator is used when you want to perform transform operations on the data columns. This processor gets automatically added on the canvas once perform any transform operation on the data columns.

 

Scheduling

To execute a workflow periodically, click on the Scheduling tab and select any of the following option to schedule.

To access the Scheduling Tab, go to the Data Pipeline home page and click on the ellipsis.

A pipeline can be scheduled:

l Minutes

l Hourly

l Daily

l Monthly

l Yearly

If Monthly is selected, then you can choose from within the options shown below:

 

Note: A pipeline can only be scheduled/unscheduled in the stopped stage.

SCHEDULE/UN-SCHEDULE: This action will un-schedule/re-schedule any sched­uled workflow and mark all the coordinated actions as killed which are in running or waiting state.

The History of the pipeline run can be viewed under the History tab:

Download/Upload Pipeline

Download Pipeline

A pipeline can be downloaded along with its component’s configurations such as Schema, Alert configuration, and Registered entities.

Upload Pipeline

A pipeline can be uploaded with its component configurations such as schema, alert definitions and registered entities.​

To upload a pipeline, click on the Upload Pipeline button on the Data Pipeline-home page.

 

If there is any duplicity in the pipeline components, a confirmation page will appear as shown below.

 

Click on Create Version button to see the Exported Pipeline Details page that pro­vides the details of the pipeline and the environment from where it was downloaded.

Click on Overwrite button and all the existing components will be overwritten with the current configuration of the pipeline, which was uploaded.

To create a new pipeline, click New Pipeline.

 

Click on Proceed, and update the pipeline configuration details on the next window. As shown in the diagram below.

The next window has two tabs, Pipeline and Component.

Pipeline button has properties such as, Pipeline Name, Log Level, Yarn Queue, and HDFS User. Expand the pipeline configurations by clicking on More button (which appears on the next screen). More button allows you to configure Error Handler Con­figuration for the modified pipeline.

Click on the Component tab to update pipeline component configuration and con­nection details if required. Once done, click on the Upload button to upload the pipe­line with the changes configured.

 

While uploading the pipeline, if the components used in the pipeline are not config­ured in the Workspace it is uploaded, then no connection will be available for that particular component.

This will not abort the upload process but the Start button of the pipeline tile will remain disabled until a connection is configured for the respective component.

Once a connection is created, update the pipeline and select a corresponding con­nection to play the pipeline.

New Pipeline button will allow to save the uploaded pipeline as a new pipeline.

First page will be the exported pipeline’s details page and after clicking on Proceed button, Pipeline configurations are displayed where the pipeline’s name and other configurations can be changed.

 

 

Stats Notification

This tab enables sending emails with the statistics of the data pipeline, to a user with customized details.

Click on Stats Notification from the pipeline tile of the Data Pipeline homepage.

This page allows you to enter the email address, on which you will receive the emails. Choose the duration of the emails, as show below:

The other tab Metrics shows all the metrics of the pipeline that will be sent in the email.

 

View Summary

Once a pipeline has been started and is in Active Mode, you can view its Summary.

To view the real-time monitoring graphs, click on the pipeline summary (go to the Data Pipeline page< Pipeline tile< click on the View Summary icon) as shown below:

To enable real time monitoring graphs, go to

Configuration >defaults >spark > EnableMonitoringGraphs and set its value to true

By default, this value is set to False.

In addition, you can view the Spark Job page from the application_1525…. link as shown below:

The View Summary tab provides the following Tabs:

Monitoring

Under the Monitoring Tab, you will find the following graphs, as shown below:

• Query Monitor

• Error Monitor

• Application Error

Query Monitor

The graph will be plotted in panels, these panels will be equal to the number of emit­ters used in the structure streaming pipeline. Each panel will have three graphs as per below discretion. These graphs can be seen through the view summary option.

Input vs. Processing Rate
The input rate specifies rate at which data is arriving from a source (Data Source) e.g. Kafka.

Processing rate specify the rate at which data from this source is being processed by Spark

Here number of input rows defines number of records processed in a trigger execu­tion.
Batch Duration
Approximate time in to process this over all micro-batch(trigger) in seconds.

Aggregation State
Shows aggregation state of current streaming query. It shows information about operator query that store the aggregation state. The graph will be seen in case if pipeline has done some aggregation.

Continuous Integration Continuous Delivery

StreamAnalytix supports continuous integration and continuous delivery of pipeline, in form of test suites and by deploying pipelines on destination.

Continuous Integration

Continuous integration is done by executing test cases (as part of a test-suite) on the incremental state of a pipeline that ensures the pipeline is not impacted by new changes

Test Suite is a collection of Test Cases associated with a pipeline.

You may want to create a test case to verify multiple permutations on continuous data which will then be deployed as soon as they pass the tests.

How to create a test case?

A test case comprises of a set of source data files (one data file for each Data Source) and inspection outputs of each component in the pipeline. These inspection outputs of each component can be thought as the expected data for the component

Every test case will have its source data associated with each Data Source. When a test suite is executed, all its associated test cases are executed too. A test suite run is uniquely identified by a run id. When a test case is executed as part of a run id, each component is inspected again, and the output data can be referred as the actual data.

How a test case is evaluated to pass or fail?

A test case is passed if both the conditions mentioned below are fulfilled:

1. Number of records are the same for actual and expected data.

2. Every field value of each record in actual data matches with the expected data.

Let’s take an example to understand this.

Your source data may have integers, or some blank values and you may want to perform certain tests to make sure that the invalid or null values are omitted. These test cases can be created with different source data files to test.

Note: When you upload a source file via Upload Data, that file is inspected, and the output is saved. However, when you use, Fetch From Source on a Data Source, StreamAnalytix internally generates a source data file using the data from the source.

To create a test case, you need to configure, save and inspect the pipeline.

Create Test Case

Once the pipeline is saved and inspected, a test case can be created. Go to the menu on the data pipeline canvas, as shown below:

 When you click on Test Suite, it opens a New Test Case window

To create a new test case, enter details in the New Test Case window fields, as shown below:

 

 

Click on Create, and the success notification message will be shown, with a new entry in the table, as shown below:

Test Suite properties

Every test case is listed in a tabular form under the components of Data pipeline can­vas.

This table has two windows (shown above), Definition and Run History with proper­ties defined below

Definition

The definition tab will show all the details and enables you to create a test case, exe­cute it, run multiple at the same time, and other actions as defined below:

Under the Coverage tab, ‘Click for more details’, will show complete configuration of the test case.

Actions on use case

Run History

You can view the test case Run History. The table will show the run statistics along with test case outputs. View the reason of failed test cases.

Stale Test Case

Multiple test cases can be configured and re-configured from the same window. Now, there may be cases when you have re-configured the pipeline and a few com­ponents have been removed.

On removing any component from existing pipeline and then creating new test case may lead to some inconsistencies to the test cases created earlier.

In case of Stale Test Cases, such inconsistencies are shown as a warning or error under the Status of the test case.

Click on Load Data again and update existing test-case to sync that test case with latest configuration of pipeline.

Each entry corresponds to a single Test Suite run. Single test suite run may contain one or more test cases.

Download Test Results

Complete test suite results can be downloaded via Download Test Results, under Results< Run History. The Test Report is a HTML file with the run Id and test run details (shown below):

 

Individual test case result can be downloaded via Download Output Results, corre­sponding to Duplicate Ids. The output result is in the JSON file format.

Updated Test Cases

If you want to update or the source file of expected result, update test case enables the same. Updating existing test case data will overwrite previous captured inspec­tion data with new inspection data. You can also update the Do Not Compare sec­tion, while editing the existing pipeline

Version Control

Version control allows you to create a version of the Test Suite. StreamAnalytix sup­ports two types of version control system:

l GIT

l StreamAnalytix Metastore

Before understanding how version control works, we need to understand what a Working Copy is.

A Working Copy is the current version of the copy or the first (0) version of the copy.

The pipeline editing is performed on the working copy, always.

Note: Any pipeline once created becomes history and cannot be edited. If you try to edit a version which is a working copy and you switch it to a previous version, it over­writes the existing working copy from selected version.

To create a version:

First way is when you re-save an edited pipeline, you get an option to Create Version.

Second way to enable version on your pipelines is when you edit and save the pipe­line. Amongst the option of saving the pipeline is Create Version and Switch version button.

Third option to create a version is by the pipeline tile.

For option 2nd and 3rd, the window show below opens.

Mention a description or a comment for the version and click on Create Version. Once you save the version, the last version on the pipeline changes from ‘Working Copy’ to Version 1.

 

A pipeline remains a working copy as long as no other version is created. Once you save a version, next to the title ‘Create Subsystem version’, is the notification of the next version that will be created.

When a new superuser logs in, they will have to configure the Version Control prop­erties under Setup.

Switch Version

If you want to switch the version of a pipeline, click on Switch Version on the pipeline editor page and choose a version. It will change the pipeline as per the selected ver­sion.

ADD SS

It is the Working Copy that is loaded to a newer version. Editing is always performed on the Working Copy of the pipeline.

NOTE: If you want to switch to an older version, save the current state as a version.

Download Version

Download Version from pipeline tile allows you to download a version of the pipe­line.

Click on the Download arrow to download the pipeline.

Continuous Delivery

While you continuously integrate data in the pipelines, you can also deliver the same.

Continuous delivery requires configuration of three environments:

l Source Environment

l Test Environment

l Destination Environment

Source environment is where the pipeline is incrementally developed.

The CD scripts download the pipeline from source environment and upload it on the Test Environment.

Then on the Test environment, test suite for the pipeline is executed. Based on user selection, a inspect session is created and then on test environment the test suite of the pipeline is executed.

If the test suite passes the execution, the pipeline is then promoted to the Destination environment.

Prerequisites

Pre-requisites for CD script

1. Installation of jq

2. SMTP server configured

Open the folder <StreamAnalytix_home>/bin/CD_script and run StartCD.sh to run the pipeline on the destination environment. Integration is done by creating Test Suites and delivery is performed by running the StartCD.sh on StreamAnalytix metastore.

Configure the below mentioned properties to deliver the test cases on the environ­ment:

Configuring Cloudera Navigator Sup­port in Data Lineage

Publishing Lineage to Cloudera Navigator

For publishing lineage to Cloudera Navigator, configure StreamAnalytix with the fol­lowing properties.

Configuration Properties

To go to the configuration properties, go To Superuser < Configuration < Others< Cloudera.

These properties are required to enable publishing of lineage to Cloudera Navigator:

 

The Navigator API Version can be extracted from the Cloudera Navigator UI. To do so, click on the question mark, located on the top right corner of the web-page, next to user name. Select ‘About’.

 

Mention the Cloudera Navigator version as the Navigator API version, as shown in the image below:

 

NOTE: The configuration ‘security.profile’ should be CDH for publishing lineage to Cloudera Navigator. This is a deployment time configuration and can be found in <StreamAnalytix_Installation_Dir>/conf/yaml/common.yaml.

Configuring Lineage in Data Pipeline

The next step to enable publishing Lineage to Cloudera Navigator is while saving the Pipeline.

Click on the checkbox shown below:

 

You can publish Lineage for both – Batch and Streaming type of pipelines in Strea­mAnalytix.

The ‘Publish lineage to Cloudera Navigator’ checkbox is enabled by default for every pipeline.

You need to update the pipeline and start.

Viewing a pipeline’s lineage on Navigator

The lineage of a pipeline is published to Cloudera Navigator once the pipeline is started.

Shown below is an example to view the lineage of the pipeline ‘sales_aggregation’.

 

Where the RabbitMQ Data Source configuration is as follows:

 

The Cloudera Navigator URL is as follows:   with credentials (usually admin/admin).

Once the pipeline is active, go to the Cloudera Navigator UI and search for the pipe­line name ‘sales_aggregation’.

 

The components of the pipeline are listed as shown above. Please note the source and target RabbitMQ components are shown as two entities in Navigator – Rabbitmq and Rabbitmq_dataset. The Rabbitmq_dataset shows the schema of the data flowing in StreamAnalytix and the Rabbitmq shows the metadata of the actual RabbitMQ component.

Click on ‘Aggregation’ entity listed above and once the 'Aggregation' entity page opens, click on ‘Lineage’ tab, as shown below:

 

Similarly, you can go to Cloudera Navigator search page and search by RabbitMQ Data Source's queue name or exchange name and view the lineage by clicking on the RabbitMQ entity.

 

You can also view the schema of the message configured on the Data Source using the ‘Details’ tab xof the RabbitMQ entity.

These are ‘Columns’ of the entity.

 

Publishing lineage to Cloudera Navigator helps you easily integrate with the existing entities.

Consider a pipeline that processes data from RabbitMQ or Kafka, enriches the data and inserts the resultant data in a hive table.

Now, an external job (not a StreamAnalytix pipeline) reads from this Hive table and does further processing of data. If lineage is published by this external job as well, you will be able to see a combined enterprise lineage on Cloudera Navigator.   

Viewing lineage of HDFS and Hive

The lineage of HDFS and Hive Data Sources and emitters are native entities in Cloud­era Navigator if the configured HDFS path or Hive table already exists.

 

The lineage of native HDFS entity is shown in green and the lineage of native Hive entity is shown in yellow.

If the HDFS path or Hive table does not exist, the lineage is represented by creating a custom dataset which is a Grey dataset.