Project Name:	CTI – Data Governance
Project ID:	173543

Document Control

Document History

Version	Date	Status <Draft/ Final>	Author	Comment / Changes from Prior Version
1.0	27-08-2019	Draft	Krupesh Shah [KS74150]	Initial Draft
1.1	02-09-2019	Draft	Krupesh Shah [KS74150]	Added Technical Approach, Data Architecture
1.2	08-09-2019	Draft	Krupesh Shah [KS74150]	Added Design Considerations. Incorporated review comments from Avishkar, Parag, Nilesh & Nagendra.
1.3	18-09-2019	Draft	Krupesh Shah [KS74150]	Enhanced Audit Design, Error File Format. Incorporated review comments from Greg.
1.4	26-09-2019	Draft	Krupesh Shah [KS74150]	Incorporated review comments from Avishkar, Parag, Nilesh & Nagendra.
2.0	27-09-2019	Baseline	Krupesh Shah [KS74150]	Technical design document is baseline and submitted for review & sign off to Greg & Carol.
3.0	29-11-2019	Baseline	Krupesh Shah [KS74150]	Modified 4.2.1 to replace the numeric and date data type check rule with Datatype_Check_Rule Modified 4.2.3 to replace value of input parameter from Master Table to Parent Table Modified 3.1.2.2 with additional description for table suffix with respective storage area like lnd, stg, mst
4.0	03-04-2020	Baseline	Krupesh Shah [KS74150]	4.2.2 Added new Integrity Rules 4.2.3 Added new Business Rules 4.2.4 Added General Rules 4.4 Added Configuration Management Information
5.0	30-12-2020	Baseline	Krupesh Shah [KS74150]	4.3.1.1 Added Continuous Ingestion Flow for API
6.0	22-01-2021	Baseline	Krupesh Shah [KS74150]	3.1.2 Technical Architecture Diagram Updated 4.4.5 Write Back Process Addition
7.0	01-03-2021	Baseline	Divya Nath [DN74106]	4.4.1.1 Continuous Ingestion updated with kafka-SQL-HIVE approach.
8.0	08-04/2021	Baseline	Hemangi Bhadke [HB40307]	4.4.1.1 Continuous Ingestion updated with kafka-HIVE-HIVE approach.

Document Reviewers/Approvers

Name	Position (e.g. Client Manager, Project Manager, Client Manager Delegate etc.)	Reviewer (only)	Reviewer and Approver
Hemangi Bhadke	Project Manager [HB40307]	X
Nilesh Sapar	Delivery Manager [NS83122]	X
Greg Book	Project Lead [GB56668]		X
Carol Pinta	Project Lead [CP02672]		X

CONTENTS

Table of Contents

1 Introduction.. 5

2 Assumptions.. 5

2.1 Technical Infrastructure Assumptions. 5

2.2 Security Assumptions. 5

2.3 Environmental Assumptions. 5

2.4 Hardware Configuration Assumptions. 5

2.5 Application Assumptions. 5

3 Technical Approach.. 6

3.1 High Level Design. 6

3.1.1 Functional Diagram.. 6

3.1.2 Technical Architecture. 7

3.2 Technology Stack. 10

4 Data Architecture.. 11

4.1 Externalized Properties file Configuration: 11

4.2 DG Connectors: 11

4.3 Rule Specifier 11

4.3.1 Schema Validation Rules. 11

4.3.2 Data Integrity Rules. 12

4.3.3 Business Rules. 13

4.3.4 General Rules. 13

4.3.5 Assumptions and Open Issues. 14

4.4 Data Flow Diagram.. 14

4.4.1 Ingestion Process. 15

4.4.1.1 Continuous Ingestion. 15

4.3.1.2 Batchwise Ingestion. 20

4.4.2 Data Integrity Rule Engine Process. 22

4.4.3 Business Rule Engine Process. 24

4.4.5 Write Back Process. 26

4.5 Configuration Management 27

4.6 Data Security. 27

4.7 Out of scope. 27

5 Use of Restricted Technology.. 27

6 Design Considerations.. 28

6.1 Deployment Plan. 28

6.1.1 Deployment Strategy. 28

6.1.2 Deployment Environment 28

6.1.3 Spark Program List 28

6.1.4 Flow List 29

6.1.5 Spark Program Execution. 29

6.1.6 Logging. 29

7 Disaster Recovery.. 29

8 Failure Recovery.. 29

9 Annexure.. 30

9.1 Ingestion Report Format 30

9.2 Data Integrity Rule Report Format 30

9.3 Business Rule Report Format 30

1 Introduction

Citi Technology Infrastructure (CTI) has 1400+ applications used for inventory management, operation management, infrastructure Support Services, etc. These applications have data entities, which are referenced across multiple applications. Most of these applications function in silos and maintain multiple copies of master data individually and using database to store. As part of this project, we would now implement Data Lake for Gold source data model.

As part of this project Data Lake solution will be developed to ingest data into new Data Lake, validate data as per Data Integrity rules & Business rules and store the Golden Source master data into HIVE. Data Ingestion and Rules validation can be scheduled depending on frequency required (Quarterly/Monthly/Daily).

2 Assumptions

2.1 Technical Infrastructure Assumptions

All the infrastructure set up will be provided by Citi infra team within Citi network.
All the specified tools will be available and running on Citi environment.

2.2 Security Assumptions

Required credentials will be shared to access the cluster.

2.3 Environmental Assumptions

We are going to use DEV Environment with Cloudera Distribution Hadoop (CDH) installed.

2.4 Hardware Configuration Assumptions

Required hardware infrastructure will be provided by Citi infra team within Citi network.
Hardware sizing will be planned based on data volume, number of rules processing, frequency of ingestions, etc.

2.5 Application Assumptions

Input feed as a File for Application data should contain the headers.
Archival tasks like clean-up of exception files, logs, etc. will be manual.
There will no UI development planned. All configurations will be done through file update.
Min and Max length are supported only for String types.
Datatype check rule is not applicable for file base Ingestion.
All configuration are generated using the Data Dictionary values and any wrong entry in the same can lead to incorrect configuration so user need to be careful while populating the DD values.

3 Technical Approach

The goal of this technical approach is to demonstrate High-Level design and document information about solution approach, Technology Architecture and tools required.

3.1 High Level Design

High-level design (HLD) explains the architecture that would be used for developing a Data Lake and provides an overview of the entire system, identifying the main components that would be developed for the product and their interfaces.

3.1.1 Functional Diagram

A functional diagram below describes Data Lake To-be functional approach which includes CTI application data sources like Database, intermediate data storage layer, rule validations, Golden Source Data Lake and reports.

3.1.2 Technical Architecture

Technology architecture below describes the technology specifications, models and guidelines using a variety of Information Technology notations.

4.3.1.2.2 Diagram

4.3.1.2.2 Technical Architecture Component Description

Item No

Name

Description

Technology

CTI Application Data Source

CTI Application Data Source is the input to Data Lake system which can be from below forms:

1. From Data File – file will be made available in Inbound directory on Data Lake

2. From CTI application System Database like SQL Server or Oracle, etc. – database connectivity will be provided from Data Lake

3. From API – Data can be captured by API connectors. E.g. Zoom Webhook

SQL Server, File, Oracle, API

Schema Generation Engine

It is the process to create the necessary tables to be created for Data Lake System.

Apache Spark, Hive

Ingestion

It is the process of storing the data from CTI application system to Data Lake distributed file system.

We will be using Apache spark for ingesting data from SQL Server or File to the Data Lake Landing Area.

Apache Spark, Hive

Landing Area

Landing Area in Data Lake is used to store the original data received from each CTI application system.

We are going to use the Hadoop Distributed File System for the storage.

It will store the day wise data and we will provide the HUE editor for querying purpose.

This will have following partition in hive:

Date: It will contain the date for which data file is there.

Sample Partition View:

– Lnd_fas_dbo_fas_fas

o 03092019

Apache Spark, Hive, HUE Editor

Data Integrity Rule Engine

Data Integrity Rule Engine will be used to execute the data integrity rules on the landing area data to identify & reject the invalid data. These rules are to be executed one column at a time to validate respective column’s data.

We will be using Spark for running the data integrity rules like not null, unique value check, min/max length etc.

Please refer section 4.2.2 for all rules.

Spark will read the data from landing area and we will apply all the rules specified in configuration on landing data and valid/invalid data will be stored to Hive according to partition strategy.

Apache Spark, Hive

Staging Area

Staging Area in Data Lake is used to store the valid data after successful execution of Data Integrity Rules.

We are going to use the Hive for the storage.

We will provide the HUE editor for querying purpose.

Apache Spark, Hive, HUE Editor

Business Rule Engine

Business Rule Engine will be used to apply the custom rules on the staging area data to validate & create the master entity related data. We will be using the spark for applying the business rule.

Please refer section 4.2.3 for all rules.

Apache Spark, Hive

Master Data Area

Master Data Area will keep the data after applying the business rules.

We will store the Master Data area in Hive.

This will contain the clean & governed data for golden source data lake.

We will provide the HUE editor for querying purpose.

This will have following partition in hive:

Date: It will contain the date for which data file is there.

Sample Partition View:

– mst_fas_dbo_fas_fas

o 03092019

Apache Spark, Hive, HUE Editor

Audit DB

This will contain the audit information about each Spark job execution.

It will have information like how many Spark jobs are executed, rule details and execution status of that rule, data time, etc. This information will be stored in Hive.

tbltAudit Table Design:

Column Name	Sample Values
Run ID	0001
Application Name	VCenter
Schema Name	Dashboard
Table Name	Clusters
Column Name	Unique_id
Path	It will come in the case of file
Count	2000
Input Parameter
Rule Name	Unique Check
Rule Description	Unique Data for table will be checked in this rule.
Status	Pass
Output Details	Out of 2000 records Unique Records are 1763.
Exception File Path	/home/{user}/Files/ExceptionFiles/11122019 /{applicationName}/{schemaName}/IntegrityReports/{filename}.csv
Execution Datetime	09/09/2019 10:10:29
CreatedBy	System
Business Date	11122019
IsNotified	1
Notified Date	09/09/2019 11:10:29

Apache Spark, Hive

Mail

Notification email will be sent to the target audience on execution of Spark Job or at specified time.

Once the email has been sent IsNotified column of tbltAudit table would be set to 1.

Sample Notification:

Run ID	Application Name	Schema Name	Table Name	Column Name	Count	Rule Name	Status	Execution Datetime	Output Details
1	Aperture	dbo_aperture	IP	VOB_ID	563	Unique Check	Pass	09/09/2019 10:10AM	563 records passed successfully

Write Back

Data can be written back from Data lake to the Database like Oracle, SQL Server

3.2 Technology Stack

Serial No	Technology Name	Purpose	Version
1	Cloudera Distribution Hadoop (CDH)	Cloudera’s open source platform distribution of Hadoop	5.14.4
2	Hadoop	The Apache Hadoop software library is a framework that allows for the distributed processing of large data sets across clusters of computers using simple programming models.	2.6.0
3	HDFS	The Hadoop Distributed File System (HDFS) is the primary data storage system used by Hadoop applications.	2.6.0
4	Apache Spark	Apache Spark is open source, general-purpose distributed computing engine used for processing and analyzing a large amount of data.	Apache Spark 2.4
5	Hive2	Hive makes data processing on Hadoop easier by providing a database query interface to Hadoop.	1.1.0
6	Hue	It provides an easy to use user interface that can be used to process all steps of Hadoop ecosystem.	3.9
7	Java	Java will be used to create application that will run on a single computer or be distributed among servers.	1.8
8	Autosys	Top of Form AutoSys is an automated job control system for scheduling, monitoring and reporting. Bottom of Form
9	Yarn	Apache Yarn is the resource management layer of Hadoop	2.6.0
10	Parquet	Apache Parquet is a columnar storage format. It allows you to lower data storage costs and maximize the effectiveness of querying data	1.5
11	NAS Drive	Network-attached storage (NAS) is a computer data storage server connected to a network.

4 Data Architecture

Data architecture is composed of file configurations, rules or standards that govern which data is collected, and how it is stored, arranged, integrated, and used in data systems and in organizations.

4.1 Externalized Properties file Configuration:

Property files will be used to store the externalized configuration e.g. Hive/database connection, log file location, log level, etc.

4.2 DG Connectors:

DG Framework will be supporting the below type of connectors:

Ingestion from SQL Server
Ingestion from Oracle
Ingestion from file [CSV, TSV, XLSX,.dat, VTX]
Conversion of VTX to CSV
Ingestion from Elastic Index
Ingestion to SQL Server from Data Lake
Ingestion to Oracle from Data Lake
Ingestion from API [JSON]
Ingestion from Kafka

4.3 Rule Specifier

Rule Specifier will contain all the schema validation, data integrity and business rules which will be implemented.

4.3.1 Schema Validation Rules

It will match the new schema with schema in configuration file and identify the discrepancy between them.

Sr.	Rule Name	Rule Description
1	Column_Check_Rule	It will identify if any new column has been added or if any column has been removed.
2	Datatype_Check_Rule	It will check the data type of source data type with data type mentioned in configuration.

These rules will be setup in a table – tblmSchemaValidationConfiguration

tblmSchemaValidationConfiguration Design:

Column Name	Sample Values
CTIApplicationName	SRM
SchemaName	SRM_MWDC
TableName	ArrayDisk
ColumnList	TIMESTAMP
ApplicableRuleName	Datatype_Check_Rule
IsDisable	N

Pseudo Code:

Schema Validation Rules engine Spark job reads all Schema Validation rules which is not disable for the specific CTI application.
It will load the table data required for each rule from the respective source and execute the rule one-by-one on those data.
One Exception File will be created for the specific CTI application (e.g. TRELLIS_DBO_TRELLIS_Schema_Validation_Check_11122019.csv) having failure details
Exception file location from the property file will be used to store the Exception file.
If schema is valid, all records will be inserted into respective tables in the Landing Area with respective partition.

4.3.2 Data Integrity Rules

These rules are to be executed one column at a time to validate respective column’s data.

Sr.	Rule Name	Rule Description
1	Unique_Check_Rule	Column should have unique values only e.g. Every master table should have one unique id column
2	Not_Null_Check_Rule	Mandatory column should be not null
3	Min_Len_Check_Rule	Column should not be less than minimum length
4	Max_Len_Check_Rule	Column should not be more than maximum length
5	Record_Count_Check_Rule	File record count should be within the tolerance limit.
6	Num_Check_Rule	Used to check that records should contain numeric value
7	DECIMAL_CHECK_RULE	Checks whether the column has decimal format or not

These rules are documented for each applications’ tables’ column in the Data Dictionary document.

These rules will be setup in a table – tblmDataIntegrityRulesConfiguration

tblmDataIntegrityRulesConfiguration Design:

Column Name	Sample Values
CTIApplicationName	SRM
SchemaName	SRM_MWDC
TableName	ArrayDisk
ColumnName	Array
ApplicableRuleName	Not_Null_Check_Rule
InputParameter
IsDisable	Y

Pseudo Code:

Data Integrity Rules engine reads all rules which are not disable for the specific CTI application.
It will load the table data required for each rule from the respective partition from the Landing Area as per the business date and execute the rules specified in configuration for those data.
Exception File will be created for the specific CTI application having failed records within that CTI application.
Exception file location from the property file will be used to store the Exception file.
All successful records will be inserted into respective tables in the Staging Area with respective partition.
All failure record will go to respective error table (original table name with suffix of _Error) and also to the single error table.

4.3.3 Business Rules

Business Rules are the custom rules to create the master entity related data.

Sr.	Rule Name	Rule Description
1	Foreign_Key_Check_Rule	Foreign key column should have data from its primary key only.
2	Join_Table_Rule	Used to combine the records of two tables into third table

These rules are documented for each applications tables’ column in the Data Dictionary document.

These rules will be setup in a table – tblmBusinessRulesConfiguration

tblmBusinessRulesConfiguration Design:

Column Name	Sample Values
CTIApplicationName	SRM
ApplicableRuleName	Foreign_Key_Check_Rule
InputParameter	ParentTable: Array\| ParentColumn :Array \| ChildTable:ArrayDisk \| ChildColumn:Array
IsDisable	N

Pseudo Code:

Business Rules engine reads all rules which are not disable for all the specific CTI applications.
It will load the table data required for each rule from the respective partition from the Staging Area as per the business date and execute the rule on those data.
Exception File will be created for all the CTI applications having failed records for all business rules.
Exception file location from the property file will be used to store the Exception file.
All successful records will be inserted into respective tables in the Master Data Area with respective partition.
All failure record will go to respective error table (original table name with suffix of _Error) and also to the single error table.

4.3.4 General Rules

These rules are internally used to record the activity in tbltAudit table.

Sr.	Rule Name	Rule Description
1	Update_next_Execution_Date_Check_Rule	Checks if next Execution date is updated.
2	Configuration_Check_rule	Checks that configuration is proper or not
3	Table_Existence_Check_Rule	Checks if table is exist or not
4	File_Existence_Check_Rule	Checks if file exist or not
5	Data_Existence_Check_Rule	Checks if data is exist or not
6	File_Conversion_Check_Rule	Check if file is converted from xlsx to csv or not
7	Path_Existence_Check_Rule	Checks if path is exist or not for file
8	File_Movement_Check_Rule	Checks if file moved from local to HDFS or not
9	Trellis_Data_Import	Checks if trellis query based data is loaded or not
10	Schema_Generation_Rule	Checks if all hive tables according to the configuration table are created successfully or not
11	Source_File_Check_Rule	Checks if the source file is read or not
12	Table_Data_Load_Check_Rule	Checks if there is successful data movement into the tables or not
13	Update Latest Execution Date	Checks if last executed date was successfuly updated in the config table or not
14	Update_Last_Execution_Date	Checks for updating table based ingestion configuration to keep track of the last time data was ingested

4.3.5 Assumptions and Open Issues

System will validate the data according to specified rules only.
Only valid records as per the rules will be ingested into the hive database.

4.4 Data Flow Diagram

A data-flow diagram (DFD) provides information about the outputs and inputs of process in the system.

4.4.1 Ingestion Process

The main ingestion process is divided into 2 parts:

4.4.1.1 Continuous Ingestion

With API
With Kafka

4.4.1.2 Batchwise Ingestion

4.4.1.1 Continuous Ingestion

4.4.1..1 Continuous Ingestion with API

API Send the continues data to the receiver. Below diagram shows the flow for Zoom which uses web hook to send the data to our framework.

The whole process is divided into 2 part:

Flow 1. Storing the data from Web Hook to NAS

Flow 2. Running the DG Framework to read the stored data from NAS and store to Hive

Flow 1. Storing the data from Web Hook to NAS

Webhook will push the data to the Web Server
DG Program is running on Web Server which will accept the data from Webhook and store it to the File on NAS

Flow 2. Running the DG Framework to read the stored data from NAS and store to Hive

Autosys will Trigger the DG Framework Job
DG Framework will take the current count and create the batch and ask the batchwise data from NAS.
NAS will send the data for specified batch
File which has been sent to DG Framework will be pushed to archive directory
DG Framework will send the data to JSON Converter
JSON converter will check the data, if it is in inclusion list it will convert to structure data and store it in hive
Autosys will trigger the retention utility job on specified time
Retention utility will remove the file from NAS archive directory

4.4.1..2 Continuous Ingestion with Kafka

Multiple sources would be publishing data into Kafka Cluster as Kafka Topics. Kafka Consumer configured on the Data Governance framework would consume these topics and parse it to finally ingest into Data Lake.

Kafka Flow

Sources will produce Topics on Kafka Bus
Multiple Producers would publish multiple topics in Kafka Cluster
Kafka Consumer / consumer groups are configured on DG framework
Kafka consumers will consume topics published on the Kafka service Bus.
Consumed Raw data goes through Ingestion framework and finally land in Master Data Lake as hive table.

Solution Approach – TBD

Presently R&D is going on 4 different approaches final approach needs to be decided.

Approach 1: For approach the consumer would place the topics as RAW data in File and parsed data is also in file.

Approach 2: For approach 2 the consumer would place the topics in Database Table and parsed file would be in file.

4.4.1..2.1 Approach 3: In this approach topics are placed in Database in RAW format and parsed data is also in Database. In this approach we are good with finding the record from the database for reconciliation purpose. Creating the table in UAT/PROD need to go from separate RLM process which can take time.

Approach 3 Design flow diagram:

Approach 3 Description:

Data Kafka topics will be decoded using base-64 decoding technique and will be consumed and stored in hive table in RAW format. Topic read and RAW data consumption is carried out using spark streaming. RAW data is then be picked up based on the topic name and further parsed using java code. This parsed data travels through DG framework for various integrity rules application if needed and is being stored in LANDING->STAGING->MASTER->MANAGED stages. MANAGED folders data is then read by Producer code written in java to produce the data back to Kafka topics to consume it by downstream systems.

This process is running every hour to read incremental data from topic and same incremental data only gets produced back. Last executed timestamp details are captured in config tables so as to read the next set of records as part of incremental process.

Flow diagram with DG framework addition can be found below:

Currently Audit & Error features are not added and TBD as the approach and design is yet to be confirmed. Audit details will be captured in NO-SQL database and discussions are in progress.

4.4.1..2.2 Approach 4: In this approach RAW file is in SQL server DB and parsing is done directly in memory in Spark, we are good with finding the record from the database for reconciliation purpose.

Parameters	Approach 1	Approach 2	Approach 3	Approach 4 (Suitable)
Parameters	File(Raw),File(Parse)	Database(Raw),File(parse)	Database(Raw),Database(Parse)	Database(Raw),Spark(parse in-memory)
Redundancy	Yes	Yes	Yes	No
Error Handling	Error Record in unparsed file	Error Record in unparsed file	Error Record in unparsed SQL table	Error Record in unparse hive table
Troubleshooting Complexity	High	High	Moderate	Moderate
Additional Development Effort	High	High	Moderate	High
Challenges	NAS mounting while EC2 Scale Up, Multiple Consumer Handling,Time Base File Rolling	Databse Purging,NAS mounting while EC2 Scale Up, Multiple Consumer Handling,Time Base File Rolling	Database Purging based on record count or volume, Additional effort required for separating from raw to multiple parse table and that table should be created already. (Check Assumption)	Database Purging based on record count or volume, On less frequency (less than 30 min) spark session required extra time
Testing Effort	Regular	High – As observed in zoom scenario need to check for record drops	Regular	Regular

Assumptions:

Topic to Table 1-1 mapping
Approach-3 – Parsed table need to create in advance with appropriate rights
Producer data sending rate and volume required to decide the frequency of DG Job

Approach 4: Details of Approach 4 is as follows:

Data for consumed topics lands into the SQL server Table in RAW format
Data value in each record is extracted after parsing.
Data is read and parsed in Spark stream
Parsed data is then ingested through DG framework
Each topic would be mapped to a table in Master Data Lake.

4.3.1.2 Batchwise Ingestion

The process describes how the data will be read from the source and stored to landing area.

4.3.1.2.2 Diagram

4.3.1.2.2 Description

Item No.

Name

Description

CTI Application Data Source

CTI Application Data Source is the input to Data Lake system which can be of two forms:

1. From Data File

2. From CTI application System Database like SQL Server or Oracle, etc.

Data Reader

This will be a spark program which will read the configured parameter make the necessary connection with database/file and read the data to data frame for further operation.

This spark program will have the following command line argument:

Command Line Argument	Description
Application Name	It will help to run program for specific application. It will be very useful when you want to run the program for only SRM again.

Table Schema Identification Process

This will be the function which will be do the following checks:

It will compare the actual column names with columns from configuration file and identify the difference like new columns added or columns which is removed.

It will read the tblmIngestionConfiguration to get all the column that should be present and it connects with database and retrieves the schema using information schema to get schema of that table. It will compare both the schema to get information of mismatched datatypes.

For file we are not comparing datatypes.

Landing Area

Once the data is loaded into data frame it will be written in day wise partitions in the landing area.

Storage details are mentioned in section 3.1.2.2.

Audit DB

This will keep the statistics of ingested data which is read successfully or failed. For Example: If we have run the rule of column check rule then Output Details of Audit DB will contain list of additional columns found if any. Audit DB details are mention in section 3.1.2.2.

tblmIngestionConfiguration Design:

Column Name	Sample Values
CTIApplicationName	SRM
SchemaName	SRM_MWDC
TableName	ArrayDisk
IsDisable	N
ColumnName	Unique_Id
ColumnDatatype	float
TargetColumnDataType	decimal(38,10)

Pseudo Code:

Data Ingestion engine Spark job reads all applicable rules which is not disable from tblmIngestionConfiguration for the specific CTI application.
It will load the table schema required for each rule and execute the rule one-by-one.
It will have the expected column and its expected data type so it will match the new schema with schema configured and identify the discrepancy between them.
One Exception File will be created for the specific CTI application having failed records for all tables’ column.
Exception file location from the property file will be used to store the Exception file
All successful records will be inserted into respective tables in the Landing Area

4.4.2 Data Integrity Rule Engine Process

The data integrity rule engine process will define the flow how data integrity rule will be applied, and result will be stored in the staging area.

4.3.1.2.2 Diagram

4.3.1.2.2 Description

Item No	Name	Description
1	Landing Area	This will contain the data which needs to be validated with the rule configured in the system. Storage details are mention in section 3.1.2.2.
2	Data Integrity Rule Engine	This will be the spark program which will read the data and rule configured from the system. Once the rule and data will be loaded into the memory it will start applying rule. Spark will read the data from landing area and it will apply all the rules specified in configuration on landing data and valid/invalid data will be stored to Hive according to partition strategy.
3	Apply Rules	This will have all the data required for rule to apply and it will apply the rule.
4	Filter Record	This process will have the valid and invalid data both so this process will be used to separate the valid and invalid records.
5	Valid Data	It will contain the valid data which will go for storage.
6	Staging Area	Once all the validation rules are applied then valid/invalid result will be stored in the Hive for further operation.
7	Invalid Data	Invalid data are those data which are failed due to rule failure. Those data will be stored in error directory structure like below : Date/Application Name/Schema Name/IntegrityReports/ SRM_MWDC_ IntegrityReports _<Date>.csv E.g. 11092019/SRM/SRM_MWDC/IntegrityReports/ SRM_MWDC_ IntegrityReports _<Date>.csv Format of Invalid File : ApplicationName, SchemaName, TableName, RuleName, FailedColumnName, FailedColumnValue, ExecutionDate
8	Audit DB	This will keep the statistics of data integrity rule. For Example: If we have run the rule of unique check then Output Details of Audit DB will contain Count of unique records. Audit DB details are mention in section 3.1.2.2.
9	Mail	Invalid data will be sent in Mail to the target audience.

4.4.3 Business Rule Engine Process

The business rule engine process will define the flow how business rule will be applied, and result will be stored in the master data area.

4.3.1.2.2 Diagram

4.3.1.2.2 Description

Item No	Name	Description
1	Staging Area	This will contain the data which needs to be validated with the rule configured in the system. Storage details are mention in section 3.1.2.2.
2	Business Rule Engine	This will be the spark program which will read the data and rule configured from the system. Once the rule and data will be loaded into the memory it will start applying rule one by one. Spark will read the data from staging area and it will apply all the rules specified in configuration on staging data and valid/invalid data will be stored to Hive according to partition strategy.
3	Apply Rules	This will have all the data required for rule to apply and it will apply the rule.
4	Filter Record	This process will have the valid and invalid data both so this process will be used to separate the valid and invalid records.
5	Valid Data	It will contain the valid data which will go for further transformation operation.
6	Master Data Area	Once all the validation rules are applied then result will be stored in the Hive.
7	Invalid Data	Invalid data are those data which are failed due to rule failure. Those data will be stored in error directory structure like below : Date/Application Name/Schema Name/BusinessReports/ SRM_MWDC_ BusinessReports _<Date>.csv E.g. 11092019/SRM/SRM_MWDC/ BusinessReports / SRM_MWDC_ BusinessReports _<Date>.csv Format of Invalid File : ApplicationName, SchemaName, TableName, RuleName, FailedColumnName, FailedColumnValue, ExecutionDate
8	Audit DB	This will keep the statistics of business rule. For Example: If we have run the rule of Record Count Check then Output Details of Audit DB will contain Count of records. Audit DB details are mention in section 3.1.2.2.
9	Mail	Invalid data will be sent in Mail to the target audience.

4.5 Configuration Management

Configuration for the system is driven based on Data Dictionary.
Once the Data Dictionary is prepared manually system runs the utility to generate the configuration for Ingestion Process, Integration Rule Engine and Business Rule Engine.
System should be able to load data to only Landing layer or to all the layer based on ISDLInsertion true or false.
All the configuration is based on Data Dictionary where system is capable to handle the history for the changes.
Data Dictionary table design is capable to see the Data Dictionary configuration for the current date or any date of history.

4.6 Data Security

Data Security will be based on existing infrastructure of Citi.

4.7 Out of scope

Data will be retained into the system till the storage limit. The data deletion, archival activity to be done on need basis as part.

5 Use of Restricted Technology

The below table describes the technology which is restricted for development:

TECHNOLOGY	Used? (Y/N)
Java Applets	N
Client-side Active X Controls	N
Other Executable Code that can be downloaded to a user/customer workstation	N
Software that requires licensing or export controls (e.g., strong encryption, third party libraries or Enterprise Resource Planning (“ERP”))	N
Streaming Media	N
Web bugs (that contain information classified as “Confidential” or “Restricted”)	N
Hidden fields (contain information classified as “Confidential” or “Restricted”)	N

6 Design Considerations

6.1 Deployment Plan

Deployment plan, i.e. defining the sequence of operations, programs or steps that will be executed manually or automatically. It consists of development strategy, development environment, programs flow which will be configured on the host system for execution.

6.1.1 Deployment Strategy

Jar will be built from development branch and then package will be pushed to environment manually.

Since we are not pushing the application to any environment other than development environment, so as of now RLM is not required.

6.1.2 Deployment Environment

Deployment environment will be based on existing Citi development environment.

Deployment environment like Hostname, Static IP and FID (Functional ID) will be provided by Citi Service Delivery team.

Deployed Hostname	Static IP	FID

6.1.3 Spark Program List

Below Spark Program lists will be configured in the flow to achieve the business requirements. There will be additional supporting programs written (like send email, write to audit, write exception file, etc.) and used as part of this flow.

Program Name	Description
Ingestion	Used to ingest data from source to Hadoop ecosystem
Integrity Rule Engine	Used to store the valid data to staging layer
Business Rule Engine	Used to apply business rule and store the clean data to Master Data Layer

6.1.4 Flow List

Flows will be configured in Autosys to run on the specified frequency.

Below is Flow list format and sample flow details.

Flow Name	Application Name	Last Execution Date	Next Execution Date	Frequency Type
SRM Execution Flow	SRM	09-09-2019	10-09-2019	Daily
VCenter Execution Flow	VCenter	09-09-2019	10-09-2019	Daily
Aperture Execution Flow	Aperture	09-09-2019	10-09-2019	Daily

Every Flow will contain collection of process (Ingestion Process, Integrity Rule Engine Process and Business Rule Engine Process) which will run.
When Flow will begin to execute, it will discover that flow next execution date is 10-09-2019 and apply rules on respective data.
It will also calculate the next execution date based on frequency type and update the Next execution date according to frequency type.

6.1.5 Spark Program Execution

Jar file for Spark program will be packaged into a separate JAR file.
Jar will be built and pushed to dev environment inside the target directory.
There will be script folder which will have the shell script which internally call the spark-submit command.
The shell script will be configured in AutoSys to execute on scheduled time.

6.1.6 Logging

Every Spark Program will be writing logs to the log file.
Log file location is mentioned in externalized properties file configuration.
Log level (info or debug or error) can be set as part of the externalized properties file configuration.
Spark program will be writing relevant information or debug statement or error details with context data related to the ongoing execution.

7 Disaster Recovery

Disaster Recovery will be based on existing infrastructure of Citi.

8 Failure Recovery

Failure Recovery will be based on existing infrastructure of Citi.

9 Annexure

9.1 Ingestion Report Format

Please find below the sample Ingestion report which can be generated from Audit DB.

Run ID

Execution Datetime

Application Name

Schema Name

Table Name

Path

Count

Input Parameter

Status

Output Details

ExceptionFilePath

Created By

Business

Date

24/09/2019 16:00:00

SRM

MWDC

Array

2000

Pass

2000 records successfully ingested

System

24092019

24/09/2019 16:00:00

SRM

MWDC

Array_Pending

5000

Pass

5000 records successfully ingested

System

24092019

24/09/2019 17:00:00

Trellis

/user/application/trellis/input

6000

Pass

6000 records successfully ingested

System

24092019

9.2 Data Integrity Rule Report Format

Please find below the sample Data Integrity Rule Report which can be generated from Audit DB.

Run ID

Execution Datetime

Application Name

Schema Name

Table Name

Column Name

Path

Count

Input Parameter

Rule Name

Rule Description

Status

Output Details

ExceptionFilePath

Created By

Business

Date

24/09/2019 16:00:00

SRM

MWDC

Array

21000

Unique_Check_Rule

Pass

CTI_DG

24092019

9.3 Business Rule Report Format

Please find below the sample Business Rule Report which can be generated from Audit DB.

Run ID

Execution Datetime

Application Name

Schema Name

Table Name

Column Name

Path

Count

Input Parameter

Rule Name

Rule Description

Status

Output Details

ExceptionFilePath

Created By

Business

Date

24/09/2019 16:00:00

SRM

MWDC

Array

21000

Foreign_Key_Check_Rule

Pass

CTI_DG

24092019

Scala Load Utility for data Produce back:

Architecture Design :

Get Destination Schema : The very first thing that the process checks if the destination connection details and schema are correct or not. There are a lot of things which are checked on later stages with the destination schema (specifically configurational columns which are not present in source).
Get Source Schema : Once the destination schema is generated, we get the source schema. Here the process checks source connection details and source schema that are given by the user are correct or not.
Generate Source Query : Once the schema and connection details of both source and destination are verified, the process further moves up in the ladder and generate dynamic query for the table name provided. This query generation process is different for each connection type (SQLServer/Hive/Oracle) and is totally dynamic with different source schemas. However if “query” parameter is given in the script, it takes precedence over dynamic query generation.
Read from Source Table : This is the next step of the script execution after we generate the source query. This query is then executed in the source environment. Point to note in this step is if the user is passing the query in query parameter he/she needs to handle the syntax of source connection type and also need to handle the “__connection_offset” column if it is present in the destination table. Some examples are included in the runbook down below for the same.
Generate Source Dataframe : After the successful execution of the query in the source environment, source dataframe is generated. This contains all the columns that are present in the source and “__connection_offset” column by default if query parameter is not given by the user. However if the destination table does not have __connection offset column in the table. It will not include the same in the query and it would only include all the columns present in the table.
Add/Remove additional columns in source dataframe : Once the source dataframe is generated process checks if the destination table contains the offset and other configuration columns which mainly include populating offset topic( this is picked up from producer config table if it is present or user can give it from topic_name parameter. Parameterized values always takes precedence over default values) , partition( this is also calculated dynamically by querying destination table before ingestion) and batchid(if it is not present in the source process adds the batchid field in the source dataframe) .
Write Back to Destination table : Once the final dataframe is generated with appropriate columns, this is finally now the time to push the data frame into the destination table. If the count of the generated dataframe is greater than 1 million (default batch size we can change the default batchsize with batchsize parameter in table) then it divides the whole load into multiple batches of 1mn ( or batch size defined in parameter ) and writes the data back to destination table. Here in case of partial load and full load the monitoring table (tbltScalaLoadMonitoring) can be checked or audit table can be checked for the status of the current process with the run id generated.
Audit and Monitoring : The whole process is full of logs and audit checks, which includes table check rule configuration check rule in case of it fails or passes. The process uses a monitoring table to keep track of the load pushed also both Info and Error logs are also gets printed on console properly to check what went wrong in case of failure or for just regular monitoring of the process.

Flow Diagram

This is going to be typical flow of a pipeline generically. For example if this script is used as producer the configurational parameters will be as shown in the image.

The whole process can be teared down in two parts

Shell script : Which is generally responsible for providing all the necessary parameters to Scala. This script generates runid, reads property file, and is responsible for passing the correct configuration including some checks.
Scala Script : Main script for ingestion, auditing and monitoring including the checks for users if a user enters the wrong information in table.

Exit codes : Currently script only works on two exit codes 0 means passed and 1 which means it is failed.

The scripts are designed in such a way that we take less inputs from user and most of the things are automated.

Runbook

Currently for configuration script uses a newly created table tblmScalaLoadConfig which has following parameters –

Parameter Name	Description	required?	Case Sensitive?
(SourceConnectionType)	This parameter defines connection type of source. Typically for now it takes 6 values only SqlServer, Oracle, Hive, MySQL, DB2 and Postgresql. It is case insensitive and a required parameter.	Yes	No
(DestinationConnectionType)	Similar to source connection type this parameter defines connection type of destination. Similar to source connection type, this also takes only 6 values SqlServer, Oracle,Hive, MySQL, DB2 and Postgresql , is case insensitive and required.	Yes	No
(SourceTableName)	As the name suggest this parameter holds the value of “fully qualified table name of source”. point to note here is this should be fully qualified with database name / schema name or the process will fail. There are no pre-defined values for this parameter and the table name check is further ahead in scala script with connection and configuration check. This parameter value is case sensitive and it is required parameter for the script.	Yes	Yes
(DestinationTableName)	Again as the name suggest this parameter holds the value of “fully qualified destination table name”. As source table name this also should be fully qualified with database name / schema name or the process will fail. Here also there are no pre-defined values for this parameter and the table name check is further ahead in scala script with connection and configuration check. similar to stn this parameter value is case sensitive and it is required parameter of the script.	Yes	Yes
(SourceUrl)	This parameter is an optional parameter in case the user doesn’t want to use the default url this parameter can be passed with appropriate connection value. Point to note here is in case the source connection type is SqlServer, this parameter becomes required as we don’t have the default SqlServer credentials in the property file. In case of sct=hive and sct=Oracle this is optional.	if sct=sqlserver, Yes else No	Yes
(SourceUsername)	As the name suggests this parameter contains the value of source username. Generally this is an optional parameter but if you have given su or sct as SqlServer this parameter becomes mandatory.	if sct=sqlserver, Yes else No	Yes
(SourcePassword)	This parameter takes password of the source connection and username passed. The password should be “decrypted” and exact as per the source username and source url. Again this is an optional parameter but if sct is given as SqlServer or su or suname is given this becomes mandate.	if sct=sqlserver, Yes else No	Yes
(DestinationUrl)	This parameter is an optional parameter in case the user doesn’t want to use the default url this parameter can be passed with appropriate connection value. Point to note here is in case the source connection type is SqlServer, this parameter becomes mandate as we don’t have the default SqlServer credentials in the property file. In case of dct=hive and dct=Oracle this is parameter is optional.	If dct=sqlserver, Yes else No	Yes
(DestinationUsername)	As the name suggests this parameter contains the value of destination username. Generally this is an optional parameter but if you have given du or dct as SqlServer this prameter becomes mandatory.	If dct=sqlserver, Yes else No	Yes
(DestinationPassword)	This parameter takes password of the destination connection and username passed. The password should be “decrypted” and exact as per the destination username and destination url. Again this is an optional parameter but if dct is given as SqlServer or du or duname is given this becomes mandate.	If dct=sqlserver, Yes else No	Yes
(Query)	This is one of the optional parameter that we have in our script. In case if less number of columns are required or some query handling is required (for e.g. if the value is ‘1’ then active and ‘0’ then inactive etc.) this query parameter is used. This is going to be most used parameter for in case of transformations on data. It is also gets appended conditions in case of incremental load.	No	Yes
(TopicName)	As name suggests, this parameter is going to contain the topic name of the current load and will be populating “__current_topic” column of the destination table. This is an optional parameter because there’s an internal handing to fetch topic name with the help of “Oracle Table Name” from tblmKafkaProducerConfig table. If the topic is not found in the producer config table this generates an error and scripts asks the user to manually enter topic name with the topic_name parameter. However this parameter takes precedence over already present topic name in producer config table (not scala load config). If both are present, parameter one will overwrite the value present in table.	if oracle table name present in producer config table No else Yes	Yes
(BatchSize)	This parameter takes only numerical value, and is responsible for holding the number of records that scala script will process in one go. This is an optional parameter and the default value of batch size is 1 mn.	No	NA
(ApplicationName)	This is the only parameter that will be given in shell script as an argument by user rest all things are handled in scala load config table. This parameter is useful in case of multiple table load under same application sequentially. If the application name is not present in the config table it’ll generate the error and redirect the user to this page.	Yes	No
(ColumnsToAdd)	Columns that is needed to be added after the final dataframe is created. For e.g. if the parameter value is “batchid=abc,source_timestamp=xyz” two columns namely batchid and sourcetimestamp will be added in df with static value for all rows as abc and xyz respectively.	No	Yes
(ColumnsToDrop)	As name suggests this parameter drops the columns from he final dataframe. For e.g. if the parameter value is “eventtimestamp,dl_createdate” both eventtimestamp and dl_createdate columns will be dropped from the final dataframe	No	Yes
(IncrementalLoadColumnName)	Name of the column that user needs to use for incremental load. Currently it is a consideration that the column must have data type as timestamp or similar data formats like date.	Yes if format is defined no otherwise	Yes
(IncrementalLoadColumnFormat)	Format of the incremental column that is mentioned in the above parameter for e.g.(“yyyy-MM-dd”, “yyyyMMdd”,”yyyy-MM-dd HH:mm:ss.SSS” etc)	Yes if column name is defined no otherwise	Yes
(LastSentTimestamp)	Stores the latest timestamp that was sent in case of incremental load. It needs to store something and can’t be empty or null. So it can be initialized with any random in format value in case of incremental load.	Yes if column name or format is defined no otherwise	NA

Some Example Commands

Hive to Oracle Load

bash scalaLoadUtilty.sh Skybox

Table parameters :

Insert query :

INSERT INTO tblmScalaLoadConfig(ApplicationName,SourceConnectionType,DestinationConnectionType,SourceTableName,DestinationTableName) VALUES (‘Skybox’,’hive’,’oracle’,’ctidsgnd1e_managed.skybox_vtm’,’TRDL_WRITE.SKYBOX_VTM_TEST’);

Few things to note here is –

Table names are fully qualified with proper schema name and database name.
Here sct and dct are case insensitive and can be passed as Hive, HIVE, hive or even HIve everything will be treated same.
stn and dtn is case sensitive and is responsibility of user.
This is full load of the table i.e all the data present in source table will be pushed to destination table.

Hive to SqlServer Load

bash scalaLoadUtilty.sh Skybox

Table parameters :

Insert query :

INSERT INTO tblmScalaLoadConfig(ApplicationName,SourceConnectionType,DestinationConnectionType,SourceTableName,DestinationTableName,DestinationUrl,DestinationUsername,DestinationPassword) VALUES (‘Skybox’,’hive’,’sqlserver’,’ctidsgnd1e_managed.skybox_vtm’,’dsg.dbo.skybox_vtm_test’,’mss-nm-7069-aol.nam.nsroot.net:2431′,’tdf’,’EncryptedPass==’);

Few things to note here is –

Destination table name is fully qualified with proper schema name and database name.
For security purpose the password is not shown in the example, but du, duname and dpwd are mandatory parameters to give as dct is sqlserver.
If any of the parameters are missing script is intelligent enough to suggest what user has missed.
Rest all points mentioned above for Hive to Oracle load is applicable for this one as well.

Oracle to SqlServer Load

bash scalaLoadUtilty.sh Skybox

Table parameters :

Insert query :

INSERT INTO tblmScalaLoadConfig(ApplicationName,SourceConnectionType,DestinationConnectionType,SourceTableName,DestinationTableName,DestinationUrl,DestinationUsername,DestinationPassword) VALUES (‘Skybox’,’oracle’,’sqlserver’,’TRDL_WRITE.SKYBOX_VTM_TEST’,’dsg.dbo.skybox_vtm_test’,’mss-nm-7069-aol.nam.nsroot.net:2431′,’tdf’,’EncryptedPass==’);

Few things to note here is –

As sct=oracle, this would take default oracle connection parameters that are fed inside the script. We need not to manually specify the connection details for oracle connection type.

SqlServer to Oracle Load

bash scalaLoadUtilty.sh Skybox

Table parameters :

Insert query :

INSERT INTO tblmScalaLoadConfig(ApplicationName,SourceConnectionType,DestinationConnectionType,SourceTableName,DestinationTableName,SourceUrl,SourceUsername,SourcePassword) VALUES (‘Skybox’,’sqlserver’,’oracle’,’dsg.dbo.skybox_vtm_test’,’TRDL_WRITE.SKYBOX_VTM_TEST’,’mss-nm-7069-aol.nam.nsroot.net:2431′,’tdf’,’EncryptedPass==’);

Few things to note here is –

As dct=oracle, this would take default oracle connection parameters that are fed inside the script. We need not to manually specify the connection details for oracle connection type.
Here as you can see order of the parameter doesn’t matter. The only thing matters is if all the required parameters are given.

Audit and Monitoring

For auditing purpose, script uses default table of DG Framework i.e. tbltAudit. There are following types of audit checks that are performed by the script in order of execution.

Check for destination table existence and configuration – This is checked first if the given configuration and table name is correct or not . If this fails script exits with exit code 1 and adds audit entry as configuration_check_rule failure. In description column we also specify what exactly is wrong table name or configuration with the exact error trace. This is also printed on console and logs.
Check for source table existence and configuration – Once destination table and configuration checks are in place then script checks for source table existence and configuration. If this fails similar to destination check script exits with exit code 1 and adds audit entry as configuration_check_rule failure. In description column we also specify what exactly is wrong table name or configuration with the exact error trace. This is also printed on console and logs.
Table data load check – Once both source and destination tables are in place the script goes ahead and tries to write the data into destination table from source table. Here if for any reason if the script is not able to write the data, this could be schema mismatch, permission issue, spark session load issue etc. If this fails with any other issue script exits with exit code 1 and adds audit entry as Table_Data_Load_Check_Rule failure. This also includes the exact count of rows that are pushed in case of partitial load. In description column we also specify what exactly is wrong table name or configuration with the exact error trace. This is also printed on console and logs.
However if the data is loaded successfully without any error a pass entry for Table_Data_Load_Check_Rule is added into audit with exact count of rows that are loaded into destination table.

Monitoring Table In DG’s SqlServer

For monitoring, script uses a newly created table tbltScalaLoadMonitoring. Monitoring entries are added as follows –

In case of failure in writing any of the records the source count would be exact but the destination count would be 0.
In case of partial load and failure, source count would be exact and destination count would be only those many number of rows which are actually written in the destination table.
In case of successful load source and destination count would be same.

Schema of Monitoring table –

Column Name	Description
RunId	Captures run id of the table to see more details in the audit table.
TopicName	null in case of __connect_topic is not present and exact topic name in case if __connect_topic is present in the destination table.
SourceConnectionType	contains exact value of source conenction type
DestinationConnectionType	Contains exact value of destination connection type
SourceTableName	Conatins exact value of source table name passed by user.
DestinationTableName	Contains exact value of destination table name passed by the user.
SourceCount	Exact count captured in source dataframe
SentCount	Exact count of rows written in destination table
StartTimestamp	Timestamp when process started
EndTimestamp	Timestamp when process ended.
ApplicationName	Name of the application which the particular table belongs to
BatchID	Exact value of count that batchid hold and we sent

API Based Ingestion :

API-based applications involve reading data from multiple source APIs, writing to landing raw table location and then parsing it further to ingesting it into hive DB.

Config :

Tablename : tblmJsonParserConfig

Flow :

Commons Failures and debugging steps:

API call failure:
- Possible Causes:
  - API is down, hence no response after hitting API.
  - Empty response from API, hence no data though got response from API.
  - Configuration not proper in tblmJsonFileParserConfig table.
  - Incorrect login credentials (status code – 401).
- Debugging Steps:
  - Check API is up by hitting API url in postman. If response is received, then API is up. Otherwise inform to source team (context team).

Syntax Highlighting: none/plain text
Expiration: Never