Data Engineering | azure-relational-data-pipeline | Malek Aldoss¶

Hi! This notebook contains part of the project.
This notebook can be foundĀ in the project's repo in two formats; .html for easy readability and .ipynb for the ability to import and make adjustments to.
The sql scripts of each view and table can be found in a .sql file format in the zip file "SqlScripts".
Purpose: The pipline copies data from 3 views into thier modeled tables.

Step 1: Create schema for new views and destination tables¶

In [ ]:
--New pipeline schema to store the views and tables used by the pipeline.
CREATE SCHEMA PipelineSchema;

Step 2: Creating three views to extract data from¶

In [9]:
--First view: To display products and product description, model, category, and culture.
CREATE VIEW [PipelineSchema].[vw_ProductDetails] 
WITH SCHEMABINDING 
AS 

SELECT 
    p.[ProductID],
    p.[Name],
    pm.[Name] AS [Model],
    pc.[Name] AS [Category],
    pmx.[Culture],
    pd.[Description] 

FROM 
    [SalesLT].[Product] p 
    INNER JOIN [SalesLT].[ProductModel] pm 
    ON p.[ProductModelID] = pm.[ProductModelID]

    INNER JOIN [SalesLT].[ProductCategory] pc 
    ON p.[ProductCategoryID] = pc.[ProductCategoryID]

    INNER JOIN [SalesLT].[ProductModelProductDescription] pmx 
    ON pm.[ProductModelID] = pmx.[ProductModelID]

    INNER JOIN [SalesLT].[ProductDescription] pd 
    ON pmx.[ProductDescriptionID] = pd.[ProductDescriptionID];
Commands completed successfully.
Total execution time: 00:00:00.049
In [4]:
--Second view: Sales orders and thier ship addresses; to analyze locations that take longer shipping times.
CREATE VIEW [PipelineSchema].[vw_ShippedOrders]
WITH SCHEMABINDING 
AS

SELECT 
    SOH.[SalesOrderID],
    SOH.[OrderDate],
    SOH.[DueDate],
    SOH.[ShipDate],
    SOH.[ShipMethod],
    A.[City],
    A.[StateProvince],
    A.[CountryRegion]

FROM 
    [SalesLT].[Address] A
    INNER JOIN [SalesLT].[SalesOrderHeader] SOH
    ON A.[AddressID] = SOH.[ShipToAddressID]

WHERE 
    SOH.[Status] = 5;
Commands completed successfully.
Total execution time: 00:00:00.049
In [11]:
--Third view: Sales personal and their sales stats to keep track of their performance.
CREATE VIEW [PipelineSchema].[vw_SalesPersonStats] 
WITH SCHEMABINDING 
AS

SELECT 
    C.[SalesPerson], 
    COUNT(C.[CustomerID]) AS TotalClients, 
    COUNT([SalesOrderID]) AS TotalSales, 
    FORMAT(SUM([SubTotal]),'C') AS TotalRevenue, 
    FORMAT(SUM([SubTotal])/COUNT([SalesOrderID]), 'C') AS AvgSale

FROM 
    [SalesLT].[SalesOrderHeader] SHO
    INNER JOIN [SalesLT].[Customer] C
    ON SHO.[CustomerID] = C.[CustomerID]

GROUP BY 
    C.[SalesPerson];
Commands completed successfully.
Total execution time: 00:00:00.048

Step 3: Creating three tables to load data in¶

In [10]:
--First Destination Table.
CREATE TABLE [PipelineSchema].[ProductDetails] (
    ProductID INT NOT NULL,
    Name NVARCHAR(256) NOT NULL,
    Model NVARCHAR(256) NOT NULL,
    Category NVARCHAR(256) NOT NULL,
    Culture NVARCHAR(6) NOT NULL,
    Description NVARCHAR(400) NOT NULL
CONSTRAINT [PK_Product_ID] PRIMARY KEY CLUSTERED ([ProductID], [Culture]) ON [PRIMARY] 
);
Commands completed successfully.
Total execution time: 00:00:00.072
In [6]:
--Second Destination Table.
CREATE TABLE [PipelineSchema].[ShippedOrders] (
    SalesOrderID INT NOT NULL,
    OrderDate DATETIME NOT NULL,
    DueDate DATETIME NOT NULL,
    ShipDate DATETIME NOT NULL,
    ShipMethod NVARCHAR(50) NOT NULL,
    City NVARCHAR(30) NOT NULL,
    StateProvince NVARCHAR(256) NOT NULL,
    CountryRegion NVARCHAR(256) NOT NULL,
CONSTRAINT [PK_Order_ID] PRIMARY KEY CLUSTERED([SalesOrderID])
);
Commands completed successfully.
Total execution time: 00:00:00.050
In [13]:
--Third Destination Table.
CREATE TABLE [PipelineSchema].[SalesPersonStats] (
    SalesPerson NVARCHAR(256) NOT NULL,
    TotalClients INT NOT NULL,
    TotalSales INT NOT NULL,
    TotalRevenue VARCHAR(256) NOT NULL,
    AvgSale VARCHAR(256) NOT NULL,
CONSTRAINT [PK_SalesPerson] PRIMARY KEY CLUSTERED([SalesPerson])
);
Commands completed successfully.
Total execution time: 00:00:00.048

Step 4: One pipeline JSON template + One support files export¶

The pipline's JSON template can be found in the zip file under the file name "pipeline_main"

The pipeline's support files can be found in the zip file under the file name "pipeline_main_support_live"

Step 5: Validate that all data records were correctly copied¶

In [1]:
--Validates all [ProductDetails] data records were moved.

--View
SELECT COUNT(*) FROM [PipelineSchema].[vw_ProductDetails];
--Table
SELECT COUNT(*) FROM [PipelineSchema].[ProductDetails];
(1 row affected)
(1 row affected)
Total execution time: 00:00:00.053
Out[1]:
(No column name)
1764
Out[1]:
(No column name)
1764
In [2]:
--Validates all [ShippedOrders] data records were moved.

--View
SELECT COUNT(*) FROM [PipelineSchema].[vw_ShippedOrders];
--Table
SELECT COUNT(*) FROM [PipelineSchema].[ShippedOrders];
(1 row affected)
(1 row affected)
Total execution time: 00:00:00.051
Out[2]:
(No column name)
32
Out[2]:
(No column name)
32
In [3]:
--Validates all [SalesPersonStats] data records were moved.

--View
SELECT COUNT(*) FROM [PipelineSchema].[vw_SalesPersonStats];
--Table
SELECT COUNT(*) FROM [PipelineSchema].[SalesPersonStats];
(1 row affected)
(1 row affected)
Total execution time: 00:00:00.051
Out[3]:
(No column name)
3
Out[3]:
(No column name)
3