Implementation<\/h3>\n
There are many patterns which can facilitate implementing the data lake but in this post we will look at creating four databases which together will handle transient application data (as acquired from the source system), staged data and finally data containing all transitional changes (inserts, deletes and updates) across all history. This design allows for a clever way of persisting all application data changes (one the core requirements of any data warehouse, irrespective of modelling approach taken) along with providing robust foundations for provisioning virtual data marts to satisfy reporting and analytical data needs. The virtual data marts can be easily implemented as views, with the required business rules embedded inside the SQL statements and, if required for performance reasons, materialised or physically persisted on disk or in memory.<\/p>\n
The following diagram depicts the high-level architecture used in this scenario and the logical placement of EDW_Landing database used for initial data load, EDW_Staging database used for storing all changes, EDW_History database storing current view of the data and finally EDW_History_Archive storing all history.<\/p>\n
To provide a tangible case scenario, let\u2019s create a sample environment with the after mentioned database using Microsoft SQL Server platform. The following script can be used to create the required data stores along with all the objects and dummy data.<\/p>\n Once executed (allowing for any environment adjustments on your side), all four databases should be present on the instance and three tables containing dummy data created on each of those. I will not go over the source to target data acquisition part as I have already covered a few different methods and frameworks which may be used for this purpose, mainly HERE<\/a> and HERE<\/a>. Therefore, assuming our data has already been copied across in its raw form into EDW_Landing database, let\u2019s focus on how to we can utilise this framework to reconcile row-level changes across all tables using this methodology.<\/p>\n This architecture can be used across many different scenarios but it is most applicable in cases where the source table does not have a reliable method to determine which records have undergone changes since the previous execution of this process, therefore making it the Data Warehouse responsibility to derive the delta. By design, this architecture accomplishes this by comparing the current source data set with the previous snapshot as maintained in the Data Lake by executing full outer join between the source data set and the equivalent copy of the data in the Data Lake (EDW_History database). Any deleted records i.e. keys which exists in EDW_History tables but do not exists in the source will be merged into the EDW_Staging dataset with the CDC operation field value set to \u2018D\u2019 (delete). The following diagram depicts the pattern used to derive all changes across source and target data sets based on hashed keys comparison (click on image to enlarge).<\/p>\n What is important to notice is the way record-level comparison is accomplished. On larger datasets e.g. over 100 million rows, comparing each value in each column across all records would yield significant performance issues. To speed up values matching across source and target data sets, a hash key is computed, concatenating all values in the tuple and hashing those to a single block of bytes using HASHBYTES() SQL Server function. Comparing hashed values in a single column (especially when indexed) makes for an easy and flexible way of comparing large datasets quickly. The following code depicts the actual implementation of this architecture, comparing EDW_Landing data with the EDW_History data and staging changed data in EDW_Staging database based on the environment and data created by the previous script.<\/p>\n Once newly arrived or changed data has been staged in EDW_Staging database we can move it to EDW_History database. As EDW_Staging database is truncated every time a load is executed, EDW_History is responsible for storing the most current version of the data along with EDW_History_Archive which is used for persisting of all previous versions of the record. As with the previous pattern of loading EDW_Staging database, history is derived based on hash keys comparison for any changed records as well as hash and primary keys comparison for new data. The following diagram depicts the change detection pattern used in populating EDW_History database.<\/p>\n![\"\"](\"http:\/\/bicortex.com\/bicortex\/wp-content\/post_content\/\/2019\/04\/DW__DataLake_Architecture_Diagram.png\")
<\/a><\/p>\n\r\n\/*========================================================================================================\r\nSTEP 1\r\nCreate EDW_Landing, EDW_Staging, EDW_History, EDW_History_Archive databases on the local instance\r\n========================================================================================================*\/\r\nSET NOCOUNT ON;\r\nGO\r\nUSE master;\r\nGO\r\nIF EXISTS (SELECT name FROM sys.databases WHERE name = N'EDW_Landing')\r\nBEGIN\r\n -- Close connections to the EDW_Landing database\r\n ALTER DATABASE EDW_Landing SET SINGLE_USER WITH ROLLBACK IMMEDIATE;\r\n DROP DATABASE EDW_Landing;\r\nEND;\r\nGO\r\n-- Create SampleDB database and log files\r\nCREATE DATABASE EDW_Landing\r\nON PRIMARY\r\n (\r\n NAME = N'EDW_Landing',\r\n FILENAME = N'D:\\Program Files\\Microsoft SQL Server\\MSSQL13.HNODWDEV\\MSSQL\\DATA\\EDW_Landing.mdf',\r\n SIZE = 10MB,\r\n MAXSIZE = 1GB,\r\n FILEGROWTH = 10MB\r\n )\r\nLOG ON\r\n (\r\n NAME = N'EDW_Landing_log',\r\n FILENAME = N'D:\\Program Files\\Microsoft SQL Server\\MSSQL13.HNODWDEV\\MSSQL\\DATA\\EDW_Landing_log.ldf',\r\n SIZE = 1MB,\r\n MAXSIZE = 1GB,\r\n FILEGROWTH = 10MB\r\n );\r\nGO\r\n--Assign database ownership to login SA\r\nEXEC EDW_Landing.dbo.sp_changedbowner @loginame = N'SA', @map = false;\r\nGO\r\n--Change the recovery model to BULK_LOGGED\r\nALTER DATABASE EDW_Landing SET RECOVERY SIMPLE;\r\nGO\r\n\r\n\r\nIF EXISTS (SELECT name FROM sys.databases WHERE name = N'EDW_Staging')\r\nBEGIN\r\n -- Close connections to the EDW_Staging database\r\n ALTER DATABASE EDW_Staging SET SINGLE_USER WITH ROLLBACK IMMEDIATE;\r\n DROP DATABASE EDW_Staging;\r\nEND;\r\nGO\r\n-- Create EDW_Staging database and log files\r\nCREATE DATABASE EDW_Staging\r\nON PRIMARY\r\n (\r\n NAME = N'EDW_Staging',\r\n FILENAME = N'D:\\Program Files\\Microsoft SQL Server\\MSSQL13.HNODWDEV\\MSSQL\\DATA\\EDW_Staging.mdf',\r\n SIZE = 10MB,\r\n MAXSIZE = 1GB,\r\n FILEGROWTH = 10MB\r\n )\r\nLOG ON\r\n (\r\n NAME = N'EDW_Staging_log',\r\n FILENAME = N'D:\\Program Files\\Microsoft SQL Server\\MSSQL13.HNODWDEV\\MSSQL\\DATA\\EDW_Staging_log.ldf',\r\n SIZE = 1MB,\r\n MAXSIZE = 1GB,\r\n FILEGROWTH = 10MB\r\n );\r\nGO\r\n--Assign database ownership to login SA\r\nEXEC EDW_Staging.dbo.sp_changedbowner @loginame = N'SA', @map = false;\r\nGO\r\n--Change the recovery model to BULK_LOGGED\r\nALTER DATABASE EDW_Staging SET RECOVERY SIMPLE;\r\nGO\r\n\r\n\r\nIF EXISTS (SELECT name FROM sys.databases WHERE name = N'EDW_History')\r\nBEGIN\r\n -- Close connections to the EDW_History database\r\n ALTER DATABASE EDW_History SET SINGLE_USER WITH ROLLBACK IMMEDIATE;\r\n DROP DATABASE EDW_History;\r\nEND;\r\nGO\r\n-- Create EDW_History database and log files\r\nCREATE DATABASE EDW_History\r\nON PRIMARY\r\n (\r\n NAME = N'EDW_History',\r\n FILENAME = N'D:\\Program Files\\Microsoft SQL Server\\MSSQL13.HNODWDEV\\MSSQL\\DATA\\EDW_History.mdf',\r\n SIZE = 10MB,\r\n MAXSIZE = 1GB,\r\n FILEGROWTH = 10MB\r\n )\r\nLOG ON\r\n (\r\n NAME = N'EDW_History_log',\r\n FILENAME = N'D:\\Program Files\\Microsoft SQL Server\\MSSQL13.HNODWDEV\\MSSQL\\DATA\\EDW_History_log.ldf',\r\n SIZE = 1MB,\r\n MAXSIZE = 1GB,\r\n FILEGROWTH = 10MB\r\n );\r\nGO\r\n--Assign database ownership to login SA\r\nEXEC EDW_History.dbo.sp_changedbowner @loginame = N'SA', @map = false;\r\nGO\r\n--Change the recovery model to BULK_LOGGED\r\nALTER DATABASE EDW_History SET RECOVERY SIMPLE;\r\nGO\r\n\r\n\r\n\r\n\r\nIF EXISTS\r\n(\r\n SELECT name\r\n FROM sys.databases\r\n WHERE name = N'EDW_History_Archive'\r\n)\r\nBEGIN\r\n -- Close connections to the EDW_History_Archive database\r\n ALTER DATABASE EDW_History_Archive\r\n SET SINGLE_USER\r\n WITH ROLLBACK IMMEDIATE;\r\n DROP DATABASE EDW_History_Archive;\r\nEND;\r\nGO\r\n-- Create EDW_History database and log files\r\nCREATE DATABASE EDW_History_Archive\r\nON PRIMARY\r\n (\r\n NAME = N'EDW_History_Archive',\r\n FILENAME = N'D:\\Program Files\\Microsoft SQL Server\\MSSQL13.HNODWDEV\\MSSQL\\DATA\\EDW_History_Archive.mdf',\r\n SIZE = 10MB,\r\n MAXSIZE = 1GB,\r\n FILEGROWTH = 10MB\r\n )\r\nLOG ON\r\n (\r\n NAME = N'EDW_History_Archive_log',\r\n FILENAME = N'D:\\Program Files\\Microsoft SQL Server\\MSSQL13.HNODWDEV\\MSSQL\\DATA\\EDW_History_Archive_log.ldf',\r\n SIZE = 1MB,\r\n MAXSIZE = 1GB,\r\n FILEGROWTH = 10MB\r\n );\r\nGO\r\n--Assign database ownership to login SA\r\nEXEC EDW_History_Archive.dbo.sp_changedbowner @loginame = N'SA',\r\n @map = false;\r\nGO\r\n--Change the recovery model to BULK_LOGGED\r\nALTER DATABASE EDW_History_Archive SET RECOVERY SIMPLE;\r\nGO\r\n\r\n\r\n\/*========================================================================================================\r\nSTEP 2\r\nCreate 'testapp' schema on all 4 databases created in Step 1\r\n========================================================================================================*\/\r\nIF OBJECT_ID('tempdb..##dbs ') IS NOT NULL\r\nBEGIN\r\n DROP TABLE ##dbs;\r\nEND;\r\nCREATE TABLE ##dbs\r\n(\r\n db VARCHAR(128),\r\n dbid SMALLINT\r\n);\r\nINSERT INTO ##dbs\r\n(\r\n db,\r\n dbid\r\n)\r\nSELECT name,\r\n database_id\r\nFROM sys.databases;\r\n\r\nEXEC dbo.sp_MSforeachdb @command1 = 'IF DB_ID(''?'') in (SELECT dbid from ##dbs where db in(''EDW_Landing'', ''EDW_Staging'',''EDW_History'', ''EDW_History_Archive'')) BEGIN USE ? EXEC(''CREATE SCHEMA testapp'') END';\r\n\r\n\r\n\r\n\/*========================================================================================================\r\nSTEP 3\r\nCreate three tables - Table1, Table2 & Table3 across all 4 databases on the testapp schema.\r\nNote: Depending on which database each table is created, its schema may vary slightly. \r\n========================================================================================================*\/\r\nUSE EDW_Landing;\r\nGO\r\nCREATE TABLE testapp.Table1\r\n(\r\n id INT NOT NULL,\r\n transaction_key UNIQUEIDENTIFIER NOT NULL,\r\n sale_price DECIMAL(10, 2) NOT NULL,\r\n customer_id INT NOT NULL,\r\n sale_datetime DATETIME NOT NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT IDENTITY(1, 1) NOT NULL,\r\n CONSTRAINT pk_testapp_table1_id\r\n PRIMARY KEY CLUSTERED (id ASC)\r\n WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON,\r\n ALLOW_PAGE_LOCKS = ON\r\n ) ON [PRIMARY]\r\n) ON [PRIMARY];\r\n\r\n\r\nUSE EDW_Landing;\r\nGO\r\nCREATE TABLE testapp.Table2\r\n(\r\n id INT NOT NULL,\r\n transaction_key UNIQUEIDENTIFIER NOT NULL,\r\n sale_price DECIMAL(10, 2) NOT NULL,\r\n customer_id INT NOT NULL,\r\n sale_datetime DATETIME NOT NULL,\r\n etl_Batch_datetime DATETIME2 NOT NULL,\r\n etl_Event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT IDENTITY(1, 1) NOT NULL,\r\n CONSTRAINT pk_testapp_table2_id\r\n PRIMARY KEY CLUSTERED (id ASC)\r\n WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON,\r\n ALLOW_PAGE_LOCKS = ON\r\n ) ON [PRIMARY]\r\n) ON [PRIMARY];\r\n\r\n\r\nUSE EDW_Landing;\r\nGO\r\nCREATE TABLE testapp.Table3\r\n(\r\n id INT NOT NULL,\r\n transaction_key UNIQUEIDENTIFIER NOT NULL,\r\n sale_price DECIMAL(10, 2) NOT NULL,\r\n customer_id INT NOT NULL,\r\n sale_datetime DATETIME NOT NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_dateTime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT IDENTITY(1, 1) NOT NULL,\r\n CONSTRAINT pk_testapp_table3_id\r\n PRIMARY KEY CLUSTERED (id ASC)\r\n WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON,\r\n ALLOW_PAGE_LOCKS = ON\r\n ) ON [PRIMARY]\r\n) ON [PRIMARY];\r\n\r\n\r\nUSE EDW_Staging;\r\nGO\r\nCREATE TABLE testapp.Table1\r\n(\r\n id INT NULL,\r\n transaction_key UNIQUEIDENTIFIER NULL,\r\n sale_price DECIMAL(10, 2) NULL,\r\n customer_id INT NULL,\r\n sale_datetime DATETIME NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT IDENTITY(1, 1) NOT NULL,\r\n etl_hash_full_record VARCHAR(512)\r\n) ON [PRIMARY];\r\n\r\n\r\nUSE EDW_Staging;\r\nGO\r\nCREATE TABLE testapp.Table2\r\n(\r\n id INT NULL,\r\n transaction_key UNIQUEIDENTIFIER NULL,\r\n sale_price DECIMAL(10, 2) NULL,\r\n customer_id INT NULL,\r\n sale_datetime DATETIME NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT IDENTITY(1, 1) NOT NULL,\r\n etl_hash_full_record VARCHAR(512)\r\n) ON [PRIMARY];\r\n\r\n\r\nUSE EDW_Staging;\r\nGO\r\nCREATE TABLE testapp.Table3\r\n(\r\n id INT NULL,\r\n transaction_key UNIQUEIDENTIFIER NULL,\r\n sale_price DECIMAL(10, 2) NULL,\r\n customer_id INT NULL,\r\n sale_datetime DATETIME NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT IDENTITY(1, 1) NOT NULL,\r\n etl_hash_full_record VARCHAR(512)\r\n) ON [PRIMARY];\r\n\r\n\r\nUSE EDW_History;\r\nGO\r\nCREATE TABLE testapp.Table1\r\n(\r\n hist_testapp_table1_sk CHAR(40) NOT NULL,\r\n id INT NULL,\r\n transaction_key UNIQUEIDENTIFIER NULL,\r\n sale_price DECIMAL(10, 2) NULL,\r\n customer_id INT NULL,\r\n sale_datetime DATETIME NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT NOT NULL,\r\n etl_hash_full_record VARCHAR(512) NOT NULL,\r\n CONSTRAINT pk_hist_testapp_table1\r\n PRIMARY KEY CLUSTERED (\r\n hist_testapp_table1_sk DESC,\r\n etl_event_datetime ASC,\r\n etl_row_id ASC\r\n )\r\n WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON,\r\n ALLOW_PAGE_LOCKS = ON, FILLFACTOR = 90\r\n ) ON [PRIMARY]\r\n) ON [PRIMARY];\r\nGO\r\n\r\n\r\nUSE EDW_History;\r\nGO\r\nCREATE TABLE testapp.Table2\r\n(\r\n hist_testapp_table2_sk CHAR(40) NOT NULL,\r\n id INT NULL,\r\n transaction_key UNIQUEIDENTIFIER NULL,\r\n sale_price DECIMAL(10, 2) NULL,\r\n customer_id INT NULL,\r\n sale_datetime DATETIME NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT NOT NULL,\r\n etl_hash_full_record VARCHAR(512) NOT NULL,\r\n CONSTRAINT pk_hist_testapp_table2\r\n PRIMARY KEY CLUSTERED (\r\n hist_testapp_table2_sk DESC,\r\n etl_event_datetime ASC,\r\n etl_row_id ASC\r\n )\r\n WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON,\r\n ALLOW_PAGE_LOCKS = ON, FILLFACTOR = 90\r\n ) ON [PRIMARY]\r\n) ON [PRIMARY];\r\nGO\r\n\r\n\r\nUSE EDW_History;\r\nGO\r\nCREATE TABLE testapp.Table3\r\n(\r\n hist_testapp_table3_sk CHAR(40) NOT NULL,\r\n id INT NULL,\r\n transaction_key UNIQUEIDENTIFIER NULL,\r\n sale_price DECIMAL(10, 2) NULL,\r\n customer_id INT NULL,\r\n sale_datetime DATETIME NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT NOT NULL,\r\n etl_hash_full_record VARCHAR(512) NOT NULL,\r\n CONSTRAINT pk_hist_testapp_table3\r\n PRIMARY KEY CLUSTERED (\r\n hist_testapp_table3_sk DESC,\r\n etl_event_datetime ASC,\r\n etl_row_id ASC\r\n )\r\n WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON,\r\n ALLOW_PAGE_LOCKS = ON, FILLFACTOR = 90\r\n ) ON [PRIMARY]\r\n) ON [PRIMARY];\r\nGO\r\n\r\n\r\nUSE EDW_History_Archive;\r\nGO\r\nCREATE TABLE testapp.Table1\r\n(\r\n hist_testapp_table1_sk CHAR(40) NOT NULL,\r\n id INT NULL,\r\n transaction_key UNIQUEIDENTIFIER NULL,\r\n sale_price DECIMAL(10, 2) NULL,\r\n customer_id INT NULL,\r\n sale_datetime DATETIME NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT NOT NULL,\r\n etl_hash_full_record VARCHAR(512) NOT NULL,\r\n CONSTRAINT pk_hist_testapp_table1\r\n PRIMARY KEY CLUSTERED (\r\n hist_testapp_table1_sk DESC,\r\n etl_event_datetime ASC,\r\n etl_row_id ASC\r\n )\r\n WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON,\r\n ALLOW_PAGE_LOCKS = ON, FILLFACTOR = 90\r\n ) ON [PRIMARY]\r\n) ON [PRIMARY];\r\nGO\r\n\r\n\r\nUSE EDW_History_Archive;\r\nGO\r\nCREATE TABLE testapp.Table2\r\n(\r\n hist_testapp_table2_sk CHAR(40) NOT NULL,\r\n id INT NULL,\r\n transaction_key UNIQUEIDENTIFIER NULL,\r\n sale_price DECIMAL(10, 2) NULL,\r\n customer_id INT NULL,\r\n sale_datetime DATETIME NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT NOT NULL,\r\n etl_hash_full_record VARCHAR(512) NOT NULL,\r\n CONSTRAINT pk_hist_testapp_table2\r\n PRIMARY KEY CLUSTERED (\r\n hist_testapp_table2_sk DESC,\r\n etl_event_datetime ASC,\r\n etl_row_id ASC\r\n )\r\n WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON,\r\n ALLOW_PAGE_LOCKS = ON, FILLFACTOR = 90\r\n ) ON [PRIMARY]\r\n) ON [PRIMARY];\r\nGO\r\n\r\n\r\nUSE EDW_History_Archive;\r\nGO\r\nCREATE TABLE testapp.Table3\r\n(\r\n hist_testapp_table3_sk CHAR(40) NOT NULL,\r\n id INT NULL,\r\n transaction_key UNIQUEIDENTIFIER NULL,\r\n sale_price DECIMAL(10, 2) NULL,\r\n customer_id INT NULL,\r\n sale_datetime DATETIME NULL,\r\n etl_batch_datetime DATETIME2 NOT NULL,\r\n etl_event_datetime DATETIME2 NOT NULL,\r\n etl_cdc_operation VARCHAR(56) NOT NULL,\r\n etl_row_id INT NOT NULL,\r\n etl_hash_full_record VARCHAR(512) NOT NULL,\r\n CONSTRAINT pk_hist_testapp_table3\r\n PRIMARY KEY CLUSTERED (\r\n hist_testapp_table3_sk DESC,\r\n etl_event_datetime ASC,\r\n etl_row_id ASC\r\n )\r\n WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON,\r\n ALLOW_PAGE_LOCKS = ON, FILLFACTOR = 90\r\n ) ON [PRIMARY]\r\n) ON [PRIMARY];\r\nGO\r\n\r\n\r\n\/*========================================================================================================\r\nSTEP 4\r\nSeed EDW_Landing database with dummy data \r\n========================================================================================================*\/\r\nUSE EDW_Landing;\r\nGO\r\nDECLARE @dt DATETIME = SYSDATETIME();\r\nDECLARE @ct INT;\r\nSET @ct = 0;\r\nWHILE @ct < 1000\r\nBEGIN\r\n INSERT INTO testapp.Table1\r\n (\r\n id,\r\n transaction_key,\r\n sale_price,\r\n customer_id,\r\n sale_datetime,\r\n etl_batch_datetime,\r\n etl_event_datetime,\r\n etl_cdc_operation\r\n )\r\n SELECT @ct + 1,\r\n CONVERT(VARCHAR(255), NEWID()),\r\n ROUND(RAND(CHECKSUM(NEWID())) * (100), 2),\r\n CAST(RAND() * 1000000 AS INT),\r\n DATEADD(DAY, (ABS(CHECKSUM(NEWID())) % 3650) * -1, GETDATE()),\r\n @dt,\r\n SYSDATETIME(),\r\n 'Insert';\r\n\r\n INSERT INTO testapp.Table2\r\n (\r\n id,\r\n transaction_key,\r\n sale_price,\r\n customer_id,\r\n sale_datetime,\r\n etl_batch_datetime,\r\n etl_event_datetime,\r\n etl_cdc_operation\r\n )\r\n SELECT @ct + 1,\r\n CONVERT(VARCHAR(255), NEWID()),\r\n ROUND(RAND(CHECKSUM(NEWID())) * (100), 2),\r\n CAST(RAND() * 1000000 AS INT),\r\n DATEADD(DAY, (ABS(CHECKSUM(NEWID())) % 3650) * -1, GETDATE()),\r\n @dt,\r\n SYSDATETIME(),\r\n 'Insert';\r\n\r\n INSERT INTO testapp.Table3\r\n (\r\n id,\r\n transaction_key,\r\n sale_price,\r\n customer_id,\r\n sale_datetime,\r\n etl_batch_datetime,\r\n etl_event_datetime,\r\n etl_cdc_operation\r\n )\r\n SELECT @ct + 1,\r\n CONVERT(VARCHAR(255), NEWID()),\r\n ROUND(RAND(CHECKSUM(NEWID())) * (100), 2),\r\n CAST(RAND() * 1000000 AS INT),\r\n DATEADD(DAY, (ABS(CHECKSUM(NEWID())) % 3650) * -1, GETDATE()),\r\n @dt,\r\n SYSDATETIME(),\r\n 'Insert';\r\n SET @ct = @ct + 1;\r\nEND;\r\n\r\n\r\nSELECT t.TABLE_CATALOG,\r\n t.TABLE_SCHEMA,\r\n t.TABLE_NAME,\r\n c.COLUMN_NAME,\r\n c.IS_NULLABLE,\r\n c.DATA_TYPE,\r\n c.ORDINAL_POSITION,\r\n c.CHARACTER_MAXIMUM_LENGTH\r\nFROM EDW_Landing.INFORMATION_SCHEMA.TABLES t\r\n JOIN EDW_Landing.INFORMATION_SCHEMA.COLUMNS c\r\n ON t.TABLE_SCHEMA = c.TABLE_SCHEMA\r\n AND t.TABLE_CATALOG = c.TABLE_CATALOG\r\n AND t.TABLE_NAME = c.TABLE_NAME\r\nWHERE c.TABLE_NAME = 'Table1';\r\n\r\nSELECT TOP 5\r\n *\r\nFROM EDW_Landing.testapp.Table1;\r\n<\/pre>\n<\/a><\/p>\n\r\n;WITH lnd\r\nAS (SELECT id,\r\n transaction_key,\r\n sale_price,\r\n customer_id,\r\n sale_datetime,\r\n etl_batch_datetime,\r\n CAST(ROW_NUMBER() OVER (PARTITION BY id ORDER BY etl_event_datetime, etl_row_id) AS INT) AS source_row_number,\r\n CONVERT(\r\n VARCHAR(40),\r\n HASHBYTES(\r\n 'SHA1',\r\n UPPER(ISNULL(RTRIM(CONVERT(VARCHAR(200), transaction_key)), 'NA') + '|'\r\n + ISNULL(RTRIM(CONVERT(VARCHAR(20), sale_price)), 'NA') + '|'\r\n + ISNULL(RTRIM(CONVERT(NVARCHAR(100), customer_id)), 'NA') + '|'\r\n + ISNULL(RTRIM(CONVERT(VARCHAR(20), id)), 'NA')\r\n + ISNULL(RTRIM(CONVERT(VARCHAR(20), etl_cdc_operation)), 'NA') + '|' + '|'\r\n )\r\n ),\r\n 2\r\n ) AS etl_hash_full_record\r\n FROM EDW_Landing.testapp.Table1),\r\n hist\r\nAS (SELECT hist.id,\r\n hist.transaction_key,\r\n hist.sale_price,\r\n hist.customer_id,\r\n hist.sale_datetime,\r\n hist.etl_hash_full_record,\r\n CONVERT(\r\n VARCHAR(40),\r\n HASHBYTES(\r\n 'SHA1',\r\n UPPER(ISNULL(RTRIM(CONVERT(VARCHAR(200), hist.transaction_key)), 'NA') + '|'\r\n + ISNULL(RTRIM(CONVERT(VARCHAR(20), hist.sale_price)), 'NA') + '|'\r\n + ISNULL(RTRIM(CONVERT(NVARCHAR(100), hist.customer_id)), 'NA')\r\n + ISNULL(RTRIM(CONVERT(VARCHAR(20), hist.id)), 'NA')\r\n + ISNULL(RTRIM(CONVERT(VARCHAR(20), 'Delete')), 'NA') + '|' + '|'\r\n )\r\n ),\r\n 2\r\n ) AS etl_hash_full_record_delete\r\n FROM EDW_History.testapp.Table1 hist\r\n JOIN\r\n (\r\n SELECT id,\r\n MAX(etl_event_datetime) AS etl_event_datetime\r\n FROM EDW_History.testapp.Table1\r\n GROUP BY id\r\n ) sub\r\n ON hist.id = sub.id\r\n AND hist.etl_event_datetime = sub.etl_event_datetime)\r\nINSERT INTO EDW_Staging.testapp.Table1\r\n(\r\n id,\r\n transaction_key,\r\n sale_price,\r\n customer_id,\r\n sale_datetime,\r\n etl_batch_datetime,\r\n etl_event_datetime,\r\n etl_cdc_operation,\r\n etl_hash_full_record\r\n)\r\nSELECT COALESCE(lnd.id, hist.id),\r\n COALESCE(lnd.transaction_key, hist.transaction_key),\r\n COALESCE(lnd.sale_price, hist.sale_price),\r\n COALESCE(lnd.customer_id, hist.customer_id),\r\n COALESCE(lnd.sale_datetime, hist.sale_datetime),\r\n COALESCE(lnd.etl_batch_datetime, GETDATE()),\r\n GETDATE(),\r\n CASE\r\n WHEN lnd.id IS NOT NULL THEN\r\n 'Insert'\r\n WHEN hist.id IS NOT NULL THEN\r\n 'Delete'\r\n ELSE\r\n NULL\r\n END,\r\n COALESCE(lnd.etl_hash_full_record, hist.etl_hash_full_record_delete)\r\nFROM lnd\r\n FULL OUTER JOIN hist\r\n ON lnd.id = hist.id\r\nWHERE hist.id IS NULL\r\n OR lnd.id IS NULL\r\n OR hist.etl_hash_full_record <> lnd.etl_hash_full_record;\r\n<\/pre>\n