ADF Pipeline beginner series : How to set a List base on parameter

 This is a series of tiny articles about some features that are lacking in the Azure data factory and how to overcome them :

To construct an array in the Azure pipeline, we need to use string parsing; in this example, we like to build a set with each element having to variable name and path; the latter comes from a pipeline parameter.

The keyword concat is very useful at it let you have multiple parameters and concatenate them into a single string. The JSON keyword will transform this into a JSON object.

@json(concat('[{"name":"/',pipeline().parameters.INITIAL_PATH,'"}]'))

A cost effective approach to build a datawarehouse in Azure Part 1:

How expensive is it to build a Data-warehouse in Azure ? 

Cloud computing provides a different paradigm while budgeting a data warehouse project. We no longer buy server but consume power. It's the same difference than buying a car or renting one by the hour. So how much power will we consume for a given project? How to reduce this consumption? 

This article is feedback after various Azure data warehouse projects. and is a compilation of services I found cost-effective.

It will details a production environment cost only (other environments depend too much of team size, corporate policies,etc..)

First, let's step back and define a warehouse

A data warehouse is a collection of business-oriented data organized per subject. These data are non-volatile and provide a single version of the truth.

Therefore we will need to accomplish three different goals :

1) Gather all meaningful and useful data

2) Store and create a history of these data, so the insight provided by reports can be challenged and audited.

3) Provide an easy way for a business user to self serve their need for analytics and reporting.

This simplified architecture shows how each Azure service interacts:

More information about this kind of architecture can be found here

 In order to provide a meaningful example, we will consider the real-life example of a medium-size company and look at sales ( sales order and invoices ) and  Stock Management. 

The volume of data to deal with is at a high level ( each area combine multiple tables ) :

Business area	Number lines	Size in GB
Products	116 000	0,156
Sales	1,6 million	1,76
Stock	3 million	2

Gather data

One of the cost-effective ways Azure have to get data is Azure Data factory leveraging Blob storage.
00
Azure Data factory can be both very expensive or very cheap depending on how you use it. The combo of dataflow and running the compute IR on Azure (Azure-hosted Integration runtime) could be insanely expensive and end up costing over 9000 USD per month to load and create a history every night just for couple millions of rows. So we will pass this idea, although from an architecture point of view this is very seductive.

Instead, we will use Data factory to load data in blob storage (which it does without breaking the bank) and organize the dataflow. Azure synapse will be used to load the data from the blob, ingest and transform these data.

This operation will cost around 4700 USD per year in our example.

Store and transform data

There are various options in Azure portfolio to store and manipulate data. First and foremost the cheapest way is pure file system storage like Blob storage, however, this is a great way to load data and gather them but fairly limited when it comes to update data and join them without expensive tools. Database on the other hand are fairly good to manipulate, update, join different datasets with a simple language. SQL.

If we look at the panel of Saas tools in the Azure repository one stand out for data warehouse, it's Azure Synapse.
It has the capability to process a large amount of data, is build for ELT scenario using blob storage as a staging area, and allow the use of multiple languages, SQL,U-SQL,Python,C#.

In our example, we use Synapse to load data from an Odata source, land them in blob storage, load them in the historical layer, organize them into Datavault artefact, use a virtual datamart to finally process them into a Tabular model. With a fair bit of margin, this is done within 5 hours every night using the least expensive option of Synapse, DWU 100 . So our ETL let us take advantage of Saas database by starting it at the very beginning of the ETL then pausing it at the very end. Therefore we are paying the database for only 5 hours. The processing power is 1,510 USD per hour therefore the cost will be 2756 USD per year + 1785 USD for 1 TB of storage per year. So all together storing and transforming the data will cost 4540 USD per year.

Provide meaningful data to business users

One way to let business access to their data is to use Azure Analysis services, this provides a model that process and add business rules so the data can be manipulated from différent tools ( Excel,Powerbi etc ..) and remain logical at all time. For example, a stock state will force a user to use a date and it will not sum up across time. Currency conversion could force the business user to choose a target currency etc...
Experience with users doing a lot of analysis with this layer translate into buying a higher level of performance. I do think B level of Azure Analysis services doesn't provide enough power. Also, bellow standard instance partitioning is not available therefore processing large table can be long as it required to reload the full dataset. Leveraging aggregate is only available with Power Bi premium ( premium include Analysis services on dedicated hardware ) but the price starts at 60 K USD per year which is a significant amount of money if the only goal is to improve performance. S1 instance provides enough power for production workload and handles a load for a medium-size company. The cost of this semantic layer is 18 k USD per year.

Product Lifecycle and CI/CD

Azure DevOps is one of the greatest tools for business intelligence in Azure. Business Intelligence is a peculiar project in IT because the data insight you deliver drive needs for more data package and deeper analysis therefore more project and work package. Therefore it's Continuous Integration/ Continuous Development tools will be a great help and Azure devops is doing that though workload management, ticket management, source control, release pipeline. Cost is rock bottom with the first 5 users free and after it's 72 USD per year per user.

Conclusion

This production environment warehouse example is the bare minimum you will need, it doesn't have any Master Data Management for example and will use excel file in blob storage or file share to let business users add mapping or reference information. Adding a good MDM software is a strong asset for any data warehouse project. However, we can have the skeleton of the data warehouse for  27 K USD per year.

A more resilient data export approach

Introduction

Business intelligence and data-warehouse are becoming a hub for communicating data to third party. Especially for regulation and legal purpose as these data need to be aggregate first before being send.These export need to be resilient in a way that an incident in the ETL or unavailability of the data warehouse should not create a gap in the data being send.

A common approach to that is subscription in SSRS or Sharepoint so the business can verify these data and the same report is send to third party. Even if some trick can let you only trigger the subscription only when the data is ready it's still lack the ability to catch up the days the ETL didn't ran or when a subscription failed to successfully send the data end to end.

Overview of the software

The software  goal is to be able to be trigger at the end of an ETL job,and create a task for each instance of a subscription for each days that need to be run. Each time a failure happen the catch up need to run as if it was when the failure happen. As a result the logic of non volatility of the data-warehouse apply to data export as well.

The software is able to send subscription via email and ftp.
The software can be found at this address here 

Detail of program behaviour:

For each active subscription the program,It will create one task per subscription for the current day.In addiction to that it also create for each subscription a task for the missing days ( So if we are the 29 and i haven't run for the 26,27,28 it will create 3 more task) .It will also create a task for each days that failed and didn't have success later one.

Each task will be execute in parallel with a max degree of parallelism set in conf file.

The flow of program is the following :

1.  Generate the list of task
2. For each task connect to Reporting services using web services,pass parameters and generate the report
3. Gather the report in memory and send it by email or ftp
4. Log the result

The catch up list all the missing day and Generate the report by passing a date to a report parameter flag as a CatchUp parameter,this parameter will receive the date of the missing/failed day with an offset.This offset is configurable.

Configure the program

Two place let you configure the program.

Application xml configuration

The first one is the SubscriptionController.exe.config file that let you configure the connection string to the report server. Endpoint address need to reflect your SSRS server. This endpoint address can be fount using SSRS Configuration manager.

The connection strings section Let you configure where your subscription database reside. The user that connect will need read and write permissions.

Within the file SubscriptionController.exe.config you will need to change the value for <web_service_ssrs_url>  <host database name> <sender_email_address(ex me@mycorp.com)> <email host (ex smtp.mycrop.com>

Database configuration

There is 8 tables :

• dbo.email - Email address for each recipient
• dbo.ftp - Host, login,password for the ftp connection
• dbo.log - Time executed,result,message,time of the report, time of the parameter
• dbo.parameter parameter name,parameter value
• dbo.report, report name,report description,report address
• dbo.subscription All subscription information
• dbo.bridge_ftp_email_subscription Link subscription to a list of email or ftp address
• dbo.bridge_parameter_subscription Link subscription to a list of parameter

More details on the subscription table :

• frequency : DAILY or WEEKLY, the weekly one will start on each Monday of the week
• type FTP or EMAIL
• active_flag, let you deactivate the subscription without dropping it, 1 mean activate 0 is deactivate
• id_report is is a FK to the report table
• type_file_export possible type are XML,csv,TIFF,PDF,MHTML,Excel,PowerPoint,Word
• title is the Subject of the email in case of an email subscription and the body as well.
• flagAddDateEndOfFile a is to add the date of the report at the end of the file
• DateFormatEndOfile is the date format for example yyyyMMdd will give four year digit two month digit and two day digits
• baseFileName is the name of the file
• dateAndOfFileOffset is the offset for the date at the end of the file, for example an offset of -1 will give set the date back by one day.
• flagEncodeASCII will encode the file in ASCII if the flag is 1
• flagCatchUpLastSuccessfullLoad if 1 it will catch on missing or failing day
• dateCatchUpOffset is the offset of the date pass to the report during a catchUp

More details on the log table :

date_execution is the date when the task have been executed

result : FAILED or SUCCESS

message : Let you know the exception in case of a failure and in case of success if it was a catch up or not

id_subscription FK to the subscription table

date_report is in case of a normal run the same as the execution date.In case of a catch up it's the date that haven't been processed  ( on earlier example 26,27,28 ).

param_date . In case of a catch up it's the actual date pass to the report for example if the date to re-process is the 26,27,28 and the offset is 2 the date pass to the report parameters will be 24,25,26.

Tracking the subscription

The report of tracking let user follow the subscription they managed with third party.This report have 5 part.

The first two,side by side on the top are a list of report parameters that are pass to the report as well as if they are catch up parameters. On the right the List of destination tell us the list of receiver,ftp or emails.

The third panel "List of subscriptions parameters" give all the parameters of the subscriptions.
Right after the fourth panel "Fail without later success".List all the instance of a subscription that failed without any later successful catch up.

Lastly the fifth panel list all the successful run letting business user know which data have been send to third party.

Historization process for datawarehouse

Introduction

Historization is a process of keeping track of data over time,it happen in different place of the data-warehouse. Today we will discuss about the one that is necessary for the audit of the input data inside the warehouse. We will call it the historization of the Staging layer and we will refer it with the HSTG acronym.

Firstly why do we need to do such a process ? We want to track everything that enter in the data-warehouse in order to understand how each data affect the insight the data-warehouse discover. We need to be able to explain figures display in reports to the final user. Source system can experience issue,to maintain the trust of user in the data-warehouse we need to explain impacts of these source system on the reports.

Overview of the process.

Logical point of view 

One of the key point of historical is to maintain an unbreakable timeline. This timeline let you query the table for a key and a date, and it will always answer back with only one row.The result row will  give the information regarding the state of this key at any point in time.The information that come back can be, this key didn't exist at this point in time,but one and only one row is always the result for any existing key.

Any table in the historical layer will have couple of extra-field providing information about history :

1. Starting point,when did this record start to be able, we will call it META_START_DATETIME
2. Ending point,when did this record stop to be relevant META_EXPIRY_DATETIME.
3. Status,( META_FLAG ) this field will tell the status of the record,I for insert ( first insert ), U for Update ( newer version of an existing record ) , D for Delete if this Business Key have been drop in the source.
4. Active flag to tell which one is the latest row META_ACTIVE_ROW,0 if its inactive 1 if its active.

In order to maintain the timeline for the beginning of time to the end of time two operations need to be perform.

First is to create a dummy row that handle the time when the record was not yet available,

So we will have a record with the exact same data than the first real record but META_START_DATETIME= 1 January 10001 at 00:00:00 am the META_EXPIRY_DATETIME is the date the actual real record become available lets said ( 9 february 2017 ). The OMD_ACTIVE_ROW.

Second the latest record for each key need to have META_EXPIRY_DATETIME set to 9999-12-31 (end of time ).

With this two operations the timeline will never be break, a query like Select * from dbo.customer where customer_key=1 and datetime = ( anydate ) will always result with one and only one  record.

Example of how data change in the HSTG depending of the source system :

Then on the second day when the address change :

On the third day the record get drop on the source system and doesn't exist anymore.

ETL point of view 

Knowing we work with 1:1 relationship between the source system and the Staging the HSTG need to do the historization the same way.As a result each table in the Staging will have a corresponding table in HSTG.

Couple step are required and they are pretty intense in term of computing power:

Full reload of the Stagging table 

Reload all the table that don't hold transaction in the Stagging layer.

Identify non-existing row in the HSTG

The first task is to create a Hash of the full row for every row in the staging, and we will compare to the hash of the full row in hstg ( minus any meta field ) that have the META_ACTIVE_ROW set to 1

We will discard ( but keep record of the number of row ) any hash that already exist.

Identify the row that are a new insert and the row that are an update

For the row who doesn't exist in the HSTG,there is two possibilities :

1. These row are a newer version of the same concept ( Business Key ) for example Customer 123 change address,we already have row for Customer 123 but the address change. So in this case the Business Key is already existing in the HSTG,these row will be tag with a U inside META_FLAG
2. The row represent a new concept ( Business Key ) which we don't have already.In this case the Business Key doesn't exist these row will be flag with an I inside the META_FLAG.

In order to deal with the deleted key the approach is going the other way than for the updated and inserted key. We taking all the hash of the HSTG and compare them to the hash of the STG.The one that don't exist are the one that have been deleted.

Conclusion

In this article we saw how we can handle historic data from a logical point of view, in order to prove and backtrack any figure coming out of the data-warehouse. In the next article we will discuss the various way to implement this within Microsoft technology. 

Generic flat file import

1) Introduction

Today we will got through a code example of how to integrate various flat file with the same generic SSIS package. Why doing such thing ? Why not just get a custom made package for each file. There is couple reason that justify this approach :

1.  Having only one package for all the data import is easier to maintain,its easier to debug if there is an issue with one file import fixing it will fix it for all the flat file import.
2. Less time to develop, why developing hundred package instead of just one ?
3. Easier to cross skill among different member of a BI team,all the file import are exactly the same,each member of the team is more likely to debug an issue.
4. Easier logic from the Scheduler point of view (Sql Agent ), all the file execute the same package,but the SSIS environment variable is different to handle the specific part of each file.
5. Easier to manage different environment,all the change are handle by SSIS environment variable 

Firstly we will go through the principals component of the package,Secondly more code details and lastly how to deploy the package.

All the file required can be find here https://bitbucket.org/vdiallonort/generic_file_import 

2) Overview of the package

For a given pattern let said All the file starting by "Customer", we will loop through all these file load them and archive these file in an separate Archive folder.

All the  success and failure and error message will be logged in the database for debugging purpose.

3) Details of each steps

1) Log the start of the process 

The first step log the start of the process it use the parameters FILE_TYPE to log what kind of file type is it ( Customer file,Sales,Invoice...),also the File Name ( Customer_20170131.txt ) the id of the row on the log table will be store on the variable LOG_ID as a result of the insert.

2) Insert the data of the CSV file into the table 

This part is where most of the work reside,

First the task will look at the structure of the file use the header to to the matching with the destination table ( parameter TABLE_DESTINATION ).

Secondly Create a rollback object and store it into an SSIS package variable. The reason to use this way of handling the rollback is to be able to rollback from C# code after any event.The rollback object is stored in a variable so it can be access from anywhere in the package.It give way more flexibility than the traditional way of starting a flow with a Begin transaction in a Sql task.

Lastly insert the data into the table.

In case of error there is an Event handler attached to this task to log any error.

3) In case of Success Move file to archive

The link between extract_and_load and the step Move to archive folder is conditional to the success of the step extract_and_load. This step is a file system task that move the current file to the Archive folder define on the parameter ARCHIVE_FILE_FOLDER. In case of success of the Move file to archive step it will go to the logging step in case of error it will go to the Rollback step ( step detailed in the 3 bis)

3 bis ) In case of failure Rollback

If the step extract_and_load fail any transaction made to the database ( except the one for logging ) will be rollback.

Its an SSIS Script task in c# with the following code :

  SqlTransaction tran = (SqlTransaction)Dts.Variables["User::transaction_object"].Value;

            tran.Rollback();

            Dts.TaskResult = (int)ScriptResults.Success;

What it does is read the SSIS variable that hold the transaction and rollback this transaction.

4) Commit all the transaction in case of success

In case both the loading and the move of the file to the archive folder where successful the transaction is committed.

5) Case of failure 

If the Rollback step have been call then the next step will log a failure of the load for this file and move to the next one.

Things to remember

There is couple things interesting to highlight,first there is one transaction per file so if a fail for any reason it doesn't affect the other ones.
The main idea is to be able to run the package twice without affecting the ETL,so this way the file is not inserted twice on the next load on a case of System error blocking the file transfer ( the file been still in the import folder it would have been pick up on the next load ) in order to achieve that if the file can be move the transaction is get rollback.

3) How to deploy

The first step is to deploy a Csv library on the server ( and any workstation that use the package ) on a folder call Library on the root of c drive,so you should have C:\Library and under two file CsvHelper.dll and CsvHelper.xml. This library is the one that read csv.

On machine after WIn7 the file CsvHelper.dll file is flag as coming from the internet causing issue later on.In order to fix it,the flag need to be remove.Go to the file properties,and unlock the file.

For the next phase you will need an existing SSIS catalogue,if you don't have any, open Integrations Services Catalogue using SSMS.

The next phase is to deploy the project using Visual Studio.First change the security option to "Do not save sensitive data".

Right click on the Project,click properties ->Deployment-> Server name and type the name of the server you want to deploy the project on.

Right click on the project and hit "Deploy",if you don't have one you will have to create a folder under the SSIS catalogue to hold your project.

Now you can find your package in the SSIS catalog architecture.

The next step is to create SSIS Environment ( one for each file type ) and map each variable inside the SSIS environment it to the package parameters. Latter on in the SSIS Job you will be able to choose which SSIS environment you want to use for this specific job.

Then we will create the variables for this environment.

Ten variables need to be create all with a String type,list bellow :

1. ARCHIVE_FILE_FOLDER This is the path of the folder holding the archive of each file
2. CS_destination                    Connection string of the destination database
3. CS_meta                              Connection string of the meta-database that store the logging information
4. FILE_PATTERN                 File pattern (for example Customer_*.csv will load all the file starting by Customer followed by any character and finishing with .csv)
5. FILE_TYPE                        File type is for logging point of view it will tell what kind of data is stored in theses set of file,for example Customer
6. SCHEMA_DESTINATION Database schema of the destination table
7. SOURCE_FILE_FOLDER Path of the folder holding the file
8. TABLE_DESTINATION Name of the destination table
9. TEXT_QUALIFIER   Text qualifier in the csv ( generally ")
10. TEXT_SEPARATOR Text separator in the csv (generally ,)

Now we will map the variable of the Package with the SSIS Environment variable.

Then for each parameter of the package to need to map it to a parameter of the SSIS environment.

So right click on the package ->Configure->References ( You can have multiple environment for the same package ).

Add the SSIS environment to the package.

Then go to Parameters and on each parameters click on the three dots.

Select the correct mapping variable.

Apply for all the parameters of the package.

You can do the same with the Connections if these change between each file it worth parameterize them.

Now create a new Job.Create a step with the package. Select SSIS package in the list of type of step.Select your package then go to configuration.Then select the corresponding SSIS environment for this file load.

Last step is to create the create the log table using this script.

CREATE TABLE [dbo].[log_file_insert](

[id] [int] IDENTITY(1,1) NOT NULL,

[file_type] [nvarchar](500) NULL,

[file_name] [nvarchar](4000) NULL,

[date_process] [datetime2](7) NULL,

[status] [int] NULL,

 CONSTRAINT [PK_log_file_insert] PRIMARY KEY CLUSTERED 

(

[id] ASC

)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON) ON [PRIMARY]

) ON [PRIMARY]

GO

DataVault and RealTime Business Intelligence

How to build Agile Business Intelligence Data Warehouse ? on top of that it need to be Real Time... 

A good modeling approach is the answer.The quality of the approach and the model is what make the difference and stabilize maintenance cost.

What does Data Vault 2.0 bring to the table ? 

- Capability to leverage unstructured  database ( Hadoop ) as traditionnl RBMS

-  Datavault 2.0 is a lego that let build your DWH brick by brick.

Why does it's interesting for RealTime BI ?

- Within a stream of data is hard and costly to join reference data and doing look-up for surrogate key is a expensive operation especially if you dealing with millions of event. By hashing Business Key it's easier to create unique key that will replace Surrogate key and avoid intensive lookup.

Let's imagine you have a flow of event representing a RFID system where a truck full of good going throw antenna that will detect them. The information that the sensor send will be a tag code,a gate code and a time stamp. 

There is couple of Business Key  we can find from this flow of data :

- First a transaction, a tag (representing a product) and a time hash together are unique ( cannot have a tag a two different place at the exact same time). and it will represent a transaction.

-  The gate code can be hash because it  represent a  Business Key

- The tag code will be hash for the same reason

- By hashing the concatenation of the tag_code,time-stamp,gate_code we represent a link between the concept of a transaction and the gate.

The second thing interesting for Real Time Business Intelligence is  you don't have to do lookup to get surrogate key.We can have a flow of transaction and in the same time a flow of Master data. The master data will go on satellite of a hub ( a hub represent a Business Concept ) and the PK of the satellite will be a hash of the Business Key+InsertTime

For example the gate will have some information about the location of the gate.

We can have a flow of data that will be :

{

   gate_code,

   gate_location_name,

   gate_longitude,

   gate_latitude

} we will hash the gate_code and obtain a gate_sk. We will be able to feed directly to the SAT_GATE. The Pk of the SAT_GATE will be gate_sk and an insert timestamp so you can have slowly changing dimension, a flag will let you know for each Business Key which row is the latest as also a start validity and the end of the validity.

Conclusion:

We can see how DataVault fit in a Real Time BI solution,by suppressing the constraint of dependency between data load.In Real Time it's harder to predict when data will arrive and you want to speed up the process as much as possible and DV2 is a great help for that.

Exemple of use of Time windows within Stream Analytics

Stream Analytics allow you to analyse your data in real-time. The really interesting feature of this engine is the capability to operate your data throw the time and create windows of time.

One really interesting windows function is the Sliding one.
 Why ?Because it allow you for each event to create a windows of time with a give size before this event. It's really interesting for sensor data where the data are raw and can be duplicate among many sensors.

On this case we will be interested for a RFID sensor that catch each tag going throw a gate,even a low speed micro controller speed up to 16 Hhz which mean you got a refresh every 62 nanoseconde. So imagine a truck full of good tagged with RFID will spend 10 second to move throw the door, the information will be send 160 millions time throw the stream of data,so instead of scale up component and have a strong power on a micro-controller just for 10 seconds we can just send this data throw the cloud and have the Event-hub scale up automatically and the stream-analytic analyse them.

So to achieve that we will be looking to set a windows function of let say 2 min ( maximum time the truck may spend to go throw the gate).

We want to discard a tag that already went throw this gate in the last 2 min.

The code on stream-analytics will be :

select gate_uid,card_uid,max(Cast([date] as datetime)) as max_datetime into outputHadoop

from input

group by gate_uid,card_uid,SlidingWindow(minute,2)

having count(*)=1;

So we grouping data by windows of 2 min sending to the output only event that not happend in the last 2 minute ( same tag at the same gate).

Conclusion : We can see it's easy to compute event using the Stream Analytics.In the case of the RFID Warehouse management system that scan in an out it's not cost-efficient to operate and purchase a strong server just for couple run a day,instead leverage the cloud power will be way more interesting, offer high scalability and speed,this point is valid for many other system that need Ad-hoc power.