Planning the Data Sources in Azure Data Lake

The goal in this section is to identify the input files (.CSV) necessary to create a consolidated SQL Azure Data Lake Analytics (ADLA) table that combines the following data sources :

1. Historical IoT Hub Streaming Data.

   1. JOIN Keys: userid, timestamp.

2. Historical IoT Hub Machine Learning Web Service Call Results.

   1. JOIN Keys: userid, timestamp.

3. Reference Data Team Member Company Affiliation.

   1. JOIN Keys: userid, id.

In the existing reference implementation, all three of these files were created in Chapters 5 and 6 when we covered the topics of Azure Stream Analytics and Azure Data Factory.

The JOIN Keys under each file represent the keys that we will use to create logical JOINs between these two tables of data in order to combine the records together and thereby enrich and enhance the data by combining these data sources together.

The JOIN with the reference data is critical so that we can match the records (by user ID) with the appropriate business entity based on the contents of the “companyname” column. We will then use this data to create individual outputs that will, in turn, be used to populate individual Azure SQL Databases by business entity.

By exposing the data as an Azure SQL database to each Business Entity, we can fully enable a “self-service” Business Intelligence (BI) experience. An additional benefit is that the Azure SQL database data is completely separated and can be further secured to reduce the legal risks of accidental data exposure to other Business Entities.

Note that in this scenario, our goal is to create a very “wide” data record that is highly denormalized to help improve the read performance. This can also enable a much easier and richer reporting experience, as there is no need for many joins on supplemental tables.

The first two files (Historical IOT Hub*.csv) were created as the result of the Stream Analytics job that persisted the incoming data streams into hot, warm, and cold data output storage paths.

The Historical IoT Hub Streaming Data files were captured as part of the “cold path” that output all of the sensor event messages sent to the IoT Hub in real time. The Stream Analytics job persisted this output directly to files in an Azure blob storage container.

The Historical IoT Hub Machine Learning Web Service Call results files were generated as part of the “hot path” that also populated a Power BI dataset for exposing real-time dashboard monitoring of team members health while on the job. This data contains all of the input parameters and associated Machine Learning (ML) Web Service Call results that were returned in real time, as each team member’s sensor data was being processed by the Azure Stream Analytics job .

The purpose of the Machine Learning Web Service calls is to predict (in real-time) if a team member was at a point of physical exhaustion or fatigue that could result in a higher chance of a work-related accident. We cover the details behind developing the Machine Learning model and exposing it as an easily consumable web service in the next chapter.

The Reference Data files were created as the result of the Azure Data Factory Job to update the Reference data, which, in turn, was used by the Azure Stream Analytics job as an “input” for reference data that could be joined to the incoming streaming data.

Overview of Tasks to Accomplish

In this section, we outline the tasks that you will need to accomplish. The goal is to populate an individual Azure SQL Database (by Business Entity) with the combined information from the previous files. These data represent the historical view of all of the sensor data recorded for a business entity’s individual team members along with the results of Machine Learning Web Service calls.

The data could be used for several purposes, including:

• Enhance Team Member Feedback: Provide historical analysis to help create an additional feedback loop (based on the physical sensor readings) to help infer and correlate “causality” surrounding daily events. Positive outcomes include noting any health-related improvement plans or behavior modifications.

• Operations Auditing: As a means of providing further operational, legal, or medical background and analysis. This data could be considered highly critical and would be analogous to a “black box” recorder: in the unlikely event that an incident should occur.

• Prediction Model Enhancement: To help provide a feedback loop to further enhance the Machine Learning prediction model for future model refinements and enhancements based on real-world data updates.

The tasks you will accomplish in the next section include the following:

1. Modify Data Factory job called CopyFromBlobToDataLake to include the three files to be moved into Data Lake.

2. Create and test new Azure Data Lake CSV Extract Scripts. Test extraction logic for all three files for proper field parsing and field type assignments.

   • Test Extract 1: IoT Hub streaming data

   • Test Extract 2: IoT Hub streaming Machine Learning Web Service results data

   • Test Extract 3: Team reference data

3. Create a new Azure Data Lake Analytics (ADLA) database and table. Populate a new table based on JOIN results from the three extract files.

4. Create separate extract files (one for each business entity) to hold the data extracted from the newly populated ADLA Database table.

5. Create separate Azure SQL databases (one for each business entity) to hold the data extracted from the ADLA database.

6. Create an Azure Data Factory job to automatically populate the Azure SQL databases based on SQL extracts (by business entity) from the ADLA database.

Now that we have outlined the plan, it’s time to get started.

Modify Azure Data Factory Job: “Copyfromblobtodatalake”

In this section, we walk through modifying the existing Data Factory pipeline named CopyFromBlobToDataLake.

Basically, we need to add two more copy activities to the pipeline in order to move these two files to Azure Data Lake:

• Historical IoT Hub Streaming Data Machine Learning Web Service Call Results from “hot” path.

• Team Member Reference Data Team Member Company Affiliation.

The easiest way to do this is to navigate to the existing Azure Data Factory Job, called CopyFromBlobToDataLake, and then select the option for Copy Data (PREVIEW) as shown in Figure 7-22.

Figure 7-22. Azure Data Factory Copy Data option

Selecting this option will allow you to easily add copy functionality to the existing Data Factory Pipeline Copy job.

Walk through the steps in the Azure data Factory Copy Data PREVIEW Wizard to create the copy job for the first additional file we need—the Historical IoT Hub Streaming Data Machine Learning Web Service Call Results. Here is a high-level outline of the input and outputs that are required to create this new copy pipeline.

New Copy Job: Historical IoT Hub Streaming Data Machine Learning Web Service Call Results:

• INPUT Dataset (Blob):

  • These files will be found in a Blob container named streamingmldata and will be generated under a folder structure of Year/Month/Day/Hour.

• OUTPUT Dataset (Data Lake):

  • Set the output data set to point to the file name of the form StreamingMLResultsDataFile{year}{month}{day}{hour}.csv.

Next, add another copy job to the existing Azure Data Factory Pipeline job to copy the second data file, the Team Member Reference Data .CSV file.

Again, we will use the Copy Data Wizard to make it fast and easy. Refer to the following input and output dataset guidance for locating and replicating your data according to the reference implementation.

Here is a high-level outline of the input and outputs that are required to create this new “copy” pipeline.

New Copy Job: Team Member Reference Data:

• INPUT Dataset (Blob):

  • These files will be found in a Blob container named refdata-team and will be generated under a file named "TeamReferenceData" + "{year}{month}{day}{hour}"+ ".csv".

  • Note that the file generations will include the extension of {year}{month}{day}{hour} to the file name.

• OUTPUT Dataset (Data Lake):

  • Set the output data set to point to the folder named streamingmldata. The file name will be "TeamReferenceData" + "{year}{month}{day}{hour}" + ".csv".

At this point, we have successfully modified our Azure Data Factory Pipeline job to copy two additional files into Azure Data Lake for building the master historical Machine Learning Analysis ADLA SQL table.

The last step is to run the updated Data Factory Copy jobs and actually move the data files from Azure Blob Storage to Azure Data Lake.

Create and Test New Azure Data Lake CSV Extract Scripts for Three Input Files

This section focuses on reading the three .CSV files we copied into Azure Data Lake from Azure Blob Storage.

In order to simplify the development experience, we will create a separate ADLA job for individually testing the extraction of each of the three files. For each file, we will simply test our parsing logic by simply reading the file and then outputting it.

To get started, navigate to your Azure Data Lake Account in the Azure Portal, and then click on the + New Job icon.

Create Extract #1: READ Streaming Data Input .CSV Files

At this point, you can name your new job “TEST EXTRACT of STREAMING DATA CSV File” and then paste the contents of this code into that file.

@streamingdata =
    EXTRACT            
            userid      string,
            age         int,
            height      int,
            weight      int,
            heartratebpm    int,
            breathingrate   double,
            temperature double,
            steps       int,
            velocity    double,
            altitude    int,
            ventilization   double,
            activity    double,
            cadence     double,
            speed       int,
            hib         double,
            heartrateredzone    double,
            heartratevariability    double,
            status     int,
            id         string,
            deviceid   string,
            messagetype int,
            longitude   double,
            latitude    double,
            timestamp   string,
            eventprocessedutctime   string,
            partitionid int,
            eventenqueuedutctime string,
            firstname   string,
            lastname    string,
            username    string,                                                                               
            utype        int,
            phone       string,
            email       string,
            gender      int,
            race        int  
    FROM "/streamingdata/2016/StreamingResultsDailyFile{*}.csv"
    USING Extractors.Csv();

OUTPUT @streamingdata  
    TO "/output/TESTStreamingResultsDailyFile20161222.csv"
    USING Outputters.Csv();

Note that the purpose of this script is for testing purposes only; it’s okay if some aspects are hard-coded. Select Save As to download a copy locally and save the script. Submit the job and check for the expected outputs. Correct any errors and re-submit the job until you have success.

When working with ADLA U-SQL scripts, it is often better to break things up into smaller chunks of code, rather than attempt to debug a larger and often more complicated script.

Make sure the first file extract is working correctly before advancing to coding the next two file extractions.

Tip Depending on how your environment is configured, you may notice quite a few .CSV files that are uniquely generated and stored under recursive folder structures.

The key to processing this type of scenario is to use a “wildcard” extension when naming the input file. In ADLA U-SQL scripts that is accomplished by using the {*} notation. For example, you can use the file name of DailySales{*}.csv to include all files that start with the name DailySales and end with a .CSV file extension.

Create Extract #2: READ Streaming ML Data Input .CSV Files

At this point, you can rename your existing job “TEST EXTRACT of STREAMING ML DATA CSV File” and then paste this code into the file .

// @streamingMLdata
@streamingmldata =
    EXTRACT            
            userid      string,
            timestamp   string,
            eventprocessedutctime   string,
            partitionid int,
            eventenqueuedutctime string,
            breathingrate   double,
            temperature double,
            ventilization string,
            activity    double,
            cadence     double,
            heartratebpm int,
            velocity    double,
            speed       int,
            hib         double,
            heartrateredzone    double,
            heartratevariability    double,
            scored_labels   string,
            scored_prob    double
    FROM "/streamingmldata/2016/StreamingMLResultsDataFile{*}.csv"
    USING Extractors.Csv();

OUTPUT @streamingmldata  
    TO "/output/TESTStreamingMLResultsDataFile20161222.csv"
    USING Outputters.Csv();

Select Save As to download a copy locally and save the script. When you are ready, submit the job and check for the expected output results. Correct any errors and re-submit the job until you can generate a complete output file.

Create Extract #3: READ Team Member Reference Data

For this job, you can rename your existing job “TEST EXTRACT of TEAM REFERENCE DATA CSV File” and then paste this code into the new file.

// @ refdatateamdata
@refdatateamdata =
    EXTRACT
        authid               string,
        companyname          string,
        firstname            string,
        lastname             string,
        username             string,
        imageUrl             string,
        utype                string,
        address1             string,
        address2             string,
        address3             string,
        city                 string,
        state                string,
        zip                  string,
        country              string,
        phone                string,
        email                string,
        linkedin             string,
        facebook             string,
        twitter              string,
        blog                 string,
        age                  int,
        height               double,
        weight               double,
        gender               int,
        race                 int,
        longitude            double,
        latitude             double,
        id                   string,
        cachettl             string,
        _rid                 string,
        _self                string,
        _etag                string,
        _attachments         string,
        _ts                  string

    FROM "/refdata-team/TeamReferenceData{*}.csv"
    USING Extractors.Csv();

OUTPUT @refdatateamdata  
    TO "/output/TESTTeamReferenceData20161222.csv"
    USING Outputters.Csv();

Select Save As to download a copy locally and save the script. When you are ready, submit the job and check for the expected output results. Correct any errors and re-submit the job until you can generate a complete output file.

At this point, you have successfully verified the three input files, and you are now ready to JOIN them to populate an ADLA database and table.

Create New Azure Data Lake Analytics (ADLA) Database and Populate New Table Based on Join Results from the Three Extracted Files

This step involves creating a new U-SQL job that will combine the three test extract scripts you created in the previous step to create a series of JOINs among the datasets. The job will then create an Azure Data Lake database, and then you will then automatically populate a new ADLA SQL table with the results of the JOIN operations .

Note that at the time of this writing, there is no UPDATE or MERGE support available in the ADLA U-SQL language for updating SQL tables. Therefore, you will usually issue DROP and CREATE U-SQL commands against your ADLA SQL tables in order to accomplish UPDATE and MERGE operations.

Note also that U-SQL provides common join operators such as INNER JOIN, LEFT/RIGHT/FULL OUTER JOIN, and SEMI JOIN to join not only ADLA SQL tables but also any row sets (even those produced from files) or external federated data sources.

For reference purposes, here are the most common types of JOIN operations and the underlying logic for matching records that they employ:

• LEFT JOIN: Return all rows from the left table and the matched rows from the right table.

• RIGHT JOIN: Return all rows from the right table and the matched rows from the left table.

• INNER JOIN: Return all rows when there is at least one match in both tables.

• FULL JOIN: Return all rows when there is a match in one of the tables.

For this job, you can rename your existing job to “EXTRACT All 3 Files and JOIN into ADLA Table”. This task results in a rather long script.

You’ll start by consolidating the three previous U-SQL extract scripts into a single script.

You will then paste the following code to implement the JOIN operations, create the ADLA database, and then output the JOIN into a new ADLA SQL table.

//(1) JOIN the streaming data with the ML Results data
@joindata1 =
    SELECT
            s.userid,      
            s.age,        
            s.height,      
            s.weight,      
            s.heartratebpm,    
            s.breathingrate,  
            s.temperature,
            s.steps,      
            s.velocity,    
            s.altitude,    
            s.ventilization,  
            s.activity,    
            s.cadence,    
            s.speed,      
            s.hib,        
            s.heartrateredzone,    
            s.heartratevariability,    
            s.status,    
            s.id,        
            s.deviceid,  
            s.messagetype,
            s.longitude,  
            s.latitude,    
            s.timestamp,  
            s.eventprocessedutctime,  
            s.partitionid,
            s.eventenqueuedutctime,
            s.firstname,  
            s.lastname,    
            s.username,    
            s.utype,        
            s.phone,      
            s.email,      
            s.gender,      
            s.race,                                                                               
            m.mluserid,
            m.mltimestamp,
            m.scoredlabels,
            m.scoredprob

    FROM @streamingdata AS s
        LEFT JOIN @streamingmldata AS m
                    ON s.timestamp == m.mltimestamp
                    AND s.userid == m.mluserid;

//(2) JOIN the NEW streaming data + ML data by USER to GET the Associated COMPANY
@joindata2 =
    SELECT
            j.userid,      
            j.age,        
            j.height,      
            j.weight,      
            j.heartratebpm,    
            j.breathingrate,  
            j.temperature,
            j.steps,      
            j.velocity,    
            j.altitude,    
            j.ventilization,  
            j.activity,    
            j.cadence,    
            j.speed,      
            j.hib,        
            j.heartrateredzone,    
            j.heartratevariability,    
            j.status,    
            j.id,        
            j.deviceid,  
            j.messagetype,
            j.longitude,  
            j.latitude,    
            j.timestamp,  
            j.eventprocessedutctime,  
            j.partitionid,
            j.eventenqueuedutctime,
            j.firstname,  
            j.lastname,    
            j.username,    
            j.utype,        
            j.phone,      
            j.email,      
            j.gender,
            j.mluserid,
            j.mltimestamp,                                                                               
            j.scoredlabels,
            j.scoredprob,      
            j.race,
            r.companyname,
            r.rid

    FROM @joindata1 AS j
       LEFT JOIN @refdatateamdata AS r
                    ON j.userid == r.rid;

OUTPUT @joindata2  
    TO "/output/SQLJoinTestResults.csv"
    USING Outputters.Csv();

DROP DATABASE IF EXISTS IOTStreamingDataMLHistoryDB;
CREATE DATABASE IOTStreamingDataMLHistoryDB;
USE DATABASE IOTStreamingDataMLHistoryDB;

CREATE TABLE StreamingDataMLHistory(    
                     INDEX sl_idx CLUSTERED (userid ASC)            
                     PARTITIONED BY HASH (userid) )
                     AS SELECT * FROM @joindata2 AS S; // Note: Automatic Table Schema generation

As before, select Save As to download a copy of the U-SQL script locally and then save the script.

When you are ready, submit the job and check for the expected output results in the ADLA SQL database. Correct any errors and re-submit the job until you can generate a successfully populated ADLA SQL table.

Note that the last statement in the U-SQL script issues a CREATE TABLE command that is populated from a SELECT statement.

The huge benefit here is that the ADLA Table Schema is implicitly inferred (column names and types) from the prior U-SQL .csv data file extraction code. This negates the need to predefine the ADLA SQL table layout before it is populated.

Create Separate Extract Files (One For Each Business Entity) to Hold the Data Extracted from the Newly Populated Azure Data Lake Analytics Database Table

In this step, you simply create separate .CSV file exports (one per business entity) from the newly populated ADLA database table.

To get started, navigate to your Azure Data Lake account in the Azure Portal, click on the +New Job icon, and then paste this code into the new job window:

//EXPORT data from ADLA Database Table to .CSV files by Business Entity

// Export #1 - WigiTech
@table1 = SELECT * FROM [IOTStreamingDataMLHistoryDB].[dbo].[StreamingDataMLHistory]
WHERE companyname =="WigiTech";

OUTPUT @table1
    TO "/output/StreamingDataMLHistory-WigiTech.Csv"
    USING Outputters.Csv();

// Export #2 - Tall Towers
@table2 = SELECT * FROM [IOTStreamingDataMLHistoryDB].[dbo].[StreamingDataMLHistory]
WHERE companyname =="Tall Towers";

OUTPUT @table2
    TO "/output/StreamingDataMLHistory-Tall-Towers.Csv"
    USING Outputters.Csv();

// Export #3 - The Complicated Badger
@table3 = SELECT * FROM [IOTStreamingDataMLHistoryDB].[dbo].[StreamingDataMLHistory]
WHERE companyname =="The Complicated Badger";

OUTPUT @table3                                                
    TO "/output/StreamingDataMLHistory-The-Complicated-Badger.Csv"
    USING Outputters.Csv();

Click on the Save As icon to save the U-SQL script. Then submit the job and check for the creation of the .CSV output result files.

Create Azure SQL Databases and Tables (One for Each Business Entity) to Hold the Data Extracted from the ADLA Database Table

The next step is to create a new Azure SQL database and SQL table to hold the information that we will extract from the ADLA SQL table that we populated from these three CSV text files:

• Historical IoT Hub Streaming Data

• Historical IoT Hub Machine Learning Web Service Call Results

• Reference Data Team Member Company Affiliation

In this exercise, you’ll perform a one-time creation of a new Azure SQL database and a new SQL table for each of the three business entities in the reference implementation.

Each of these databases will contain the merged dataset results (for one of the business entities), where you merged the three files into an ADLA SQL table.

To complete this task, you need the following prerequisites:

• Azure SQL Server Database information, such as server, user, and password credentials.

• Azure SQL Server firewall ports opened-up to allow your client IP address to information.

• SQL Server Management Studio (SSMS) or Visual Studio Server Explorer to run the Transact-SQL scripts to create the three Azure SQL databases and tables.

Note Download SQL Server Management Studio (SSMS): https://msdn.microsoft.com/en-us/library/mt238290.aspx .

Once you have established connectivity to your Azure SQL Server database using either SQL Server Management Studio or Visual Studio Server Explorer, open a new Query window and enter this SQL script:

USE [master]
GO

DROP DATABASE IF EXISTS [IOTDataMLHistory-WigiTech]
GO

CREATE DATABASE [IOTDataMLHistory-WigiTech]
GO

USE [IOTDataMLHistory-WigiTech]
GO

DROP TABLE  IF EXISTS [dbo].[IOTDataMLHistory]
GO

CREATE TABLE [dbo].[IOTDataMLHistory](
     [UserId] [char](256) NOT NULL,
     [Age] [float] NOT NULL,
     [Height] [float] NOT NULL,
     [Weight] [float] NOT NULL,
     [HeartRateBPM] [float] NOT NULL,
     [BreathingRate] [float] NOT NULL,
     [Temperature] [float] NOT NULL,
     [Steps] [float] NOT NULL,
     [Velocity] [float] NOT NULL,
     [Altitude] [float] NOT NULL,
     [Ventilization] [float] NOT NULL,
     [Activity] [float] NOT NULL,
     [Cadence] [float] NOT NULL,
     [Speed] [float] NOT NULL,
     [HIB] [float] NOT NULL,
     [HeartRateRedZone] [float] NOT NULL,
     [HeartrateVariability] [float] NOT NULL,
     [Status] [int] NOT NULL,
     [Id] [char](256) NOT NULL,
     [DeviceId] [char](256) NOT NULL,
     [MessageType] [int] NOT NULL,                                                                               
     [Longitude] [float] NOT NULL,
     [Latitude] [float] NOT NULL,
     [Timestamp] [datetime2](7) NOT NULL,
     [EventProcessedUtcTime] [datetime2](7) NOT NULL,
     [PartitionId] [int] NOT NULL,
     [EventEnqueuedUtcTime] [datetime2](7) NOT NULL,
     [FirstName]  [char](256) NOT NULL,
     [LastName]  [char](256) NOT NULL,
     [UserName]  [char](256) NOT NULL,
     [UType]  [char](256) NOT NULL,
     [Phone]  [char](256) NOT NULL,
     [Email]  [char](256) NOT NULL,
     [Gender]  [char](256) NOT NULL,
     [MLUserid]  [char](256) NOT NULL,
     [MLTimestamp]  [char](256) NOT NULL,
     [ScoredLabels]  [char](256) NOT NULL,
     [ScoredProb]  [float] NOT NULL,
     [Race]   [int] NOT NULL,
     [CompanyName]  [char](256) NOT NULL,
     [Rid]  [char](256) NOT NULL
)
GO

After running this script three times (modifying it to reflect each business entity), you will be ready to populate the Azure SQL database tables by using an Azure Data Factory job.

Create Azure Data Factory Job to Automatically Populate the Azure Sql Databases (By Business Entity) Using the Sql Query Extractions from the Azure Data Lake Database

This last step will complete the data analysis and reporting journey by exposing the historical IoT sensor data and associated Machine Learning Web Service call results for consumption by the business entities in the reference implementation scenario.

At this point, all the hard work has already been completed, and all you have to do is create an Azure Data Factory job to select the data out of the ADLA SQL database, filter the results by business entity, and then insert the data into the individual Azure SQL databases you created in the last step.

To get started via the Azure Portal, first navigate to your existing Azure resource group. Search for Data Factory and then add a new Data Factory job, as shown in Figure 7-23.

Figure 7-23. Adding a new Azure Data Factory job to a resource group

Next, complete the Azure Data Factory Job configuration parameters, as shown in Figure 7-24.

Figure 7-24. Azure Data Factory job parameters

Click on the Create icon, and then the new Data Factory job will be provisioned into your Azure Resource Group after a few minutes.

After the new Data Factory job has been provisioned into your environment, navigate to the new Data Factory job and select the Copy Data (PREVIEW) icon to launch the Copy Wizard.

Figure 7-25. Data Factory Copy Data Wizard

Next, walk through the Copy Data Wizard to create a Data Factory pipeline job that will copy the data from each individual .CSV file (one for each entity) to the appropriate Azure SQL database (which varies by business entity). The SQL table name and scheme in each database is the same across all databases.

Here are code fragments from a sample Azure Data Factory job to populate an Azure SQL Database for WigiTech, based on a .CSV file exported from the Azure Data Lake Store .

Linked Service: Source Azure Data Lake Store

{
    "name": "Source-DataLakeStore-f0g",
    "properties": {
        "hubName": "populateazuresqldbs_hub",
        "type": "AzureDataLakeStore",
        "typeProperties": {
            "dataLakeStoreUri": "https://brtadls.azuredatalakestore.net/webhdfs/v1",
            "authorization": "**********",
            "sessionId": "**********",
            "subscriptionId": "<Your Subscription ID>",
            "resourceGroupName": "brtwhdev"
        }
    }
}

Linked Service: Destination: Azure SQL Database

{
    "name": "Destination-SQLAzure-f0g",
    "properties": {
        "hubName": "populateazuresqldbs_hub",
        "type": "AzureSqlDatabase",
        "typeProperties": {
            "connectionString": "Data Source=brtsqlserverdev.database.windows.net;Initial Catalog=IOTDataMLHistory-WigiTech;Integrated Security=False;User ID=<Your User  ID>;Password=<Your Password>;Connect Timeout=30;Encrypt=True"
        }
    }
}                                                                                                    

Input Dataset : .CSV File in Azure Data Lake Store

{
    "name": "InputDataset-f0g",                                                        
    "properties": {
        "structure": [
            {
                "name": "Column0",
                "type": "String"
            },
            {
                "name": "Column1",
                "type": "Int64"
            },
            {
                "name": "Column2",
                "type": "Int64"
            },
            {
                "name": "Column3",
                "type": "Int64"
            },
            {
                "name": "Column4",
                "type": "Int64"
            },
            {
                "name": "Column5",
                "type": "Double"
            },
            {
                "name": "Column6",
                "type": "Double"
            },
            {
                "name": "Column7",
                "type": "Int64"
            },
            {
                "name": "Column8",
                "type": "Double"                                                        
            },
            {
                "name": "Column9",
                "type": "Int64"
            },
            {
                "name": "Column10",                                                                                                    
                "type": "Double"
            },
            {
                "name": "Column11",
                "type": "Double"
            },
            {
                "name": "Column12",
                "type": "Int64"
            },
            {
                "name": "Column13",
                "type": "Int64"
            },
            {
                "name": "Column14",
                "type": "Int64"
            },
            {
                "name": "Column15",
                "type": "Int64"
            },
            {
                "name": "Column16",
                "type": "Int64"
            },
            {
                "name": "Column17",
                "type": "Int64"
            },
            {
                "name": "Column18",
                "type": "String"
            },
            {
                "name": "Column19",
                "type": "String"
            },
            {
                "name": "Column20",
                "type": "Int64"
            },
            {
                "name": "Column21",
                "type": "Double"                                                                                                    
            },
            {
                "name": "Column22",
                "type": "Double"
            },
            {
                "name": "Column23",
                "type": "Datetime"
            },
            {
                "name": "Column24",
                "type": "Datetime"
            },
            {
                "name": "Column25",
                "type": "Int64"
            },
            {
                "name": "Column26",
                "type": "Datetime"
            },
            {
                "name": "Column27",
                "type": "String"
            },
            {
                "name": "Column28",
                "type": "String"
            },
            {
                "name": "Column29",
                "type": "String"
            },
            {
                "name": "Column30",
                "type": "Int64"
            },
            {
                "name": "Column31",
                "type": "Int64"
            },
            {
                "name": "Column32",
                "type": "String"
            },                                                                                                    
            {
                "name": "Column33",
                "type": "Int64"
            },
            {
                "name": "Column34",
                "type": "String"
            },
            {
                "name": "Column35",
                "type": "Datetime"
            },
            {
                "name": "Column36",
                "type": "String"
            },
            {
                "name": "Column37",
                "type": "Double"
            },
            {
                "name": "Column38",
                "type": "Int64"
            },
            {
                "name": "Column39",
                "type": "String"
            },
            {
                "name": "Column40",
                "type": "String"
            }
        ],
        "published": false,
        "type": "AzureDataLakeStore",
        "linkedServiceName": "Source-DataLakeStore-f0g",
        "typeProperties": {
            "fileName": "StreamingDataMLHistory-WigiTech.Csv",
            "folderPath": "output/",
            "format": {
                "type": "TextFormat",
                "columnDelimiter": ",",
                "quoteChar": """
            }                                                                                                    
        },
        "availability": {
            "frequency": "Day",
            "interval": 1
        },
        "external": true,
        "policy": {}
    }
}

Output Dataset : Azure SQL Database Table

{
    "name": "OutputDataset-f0g",
    "properties": {
        "structure": [
            {
                "name": "UserId",
                "type": "String"
            },
            {
                "name": "Age",
                "type": "Double"
            },
            {
                "name": "Height",
                "type": "Double"
            },
            {
                "name": "Weight",
                "type": "Double"
            },
            {
                "name": "HeartRateBPM",
                "type": "Double"
            },
            {
                "name": "BreathingRate",
                "type": "Double"
            },
            {
                "name": "Temperature",
                "type": "Double"
            },
            {                                                                                                    
                "name": "Steps",
                "type": "Double"
            },
            {
                "name": "Velocity",
                "type": "Double"
            },
            {
                "name": "Altitude",
                "type": "Double"
            },
            {
                "name": "Ventilization",
                "type": "Double"
            },
            {
                "name": "Activity",
                "type": "Double"
            },
            {
                "name": "Cadence",
                "type": "Double"
            },
            {
                "name": "Speed",
                "type": "Double"
            },
            {
                "name": "HIB",
                "type": "Double"
            },
            {
                "name": "HeartRateRedZone",
                "type": "Double"
            },
            {
                "name": "HeartrateVariability",
                "type": "Double"
            },
            {
                "name": "Status",
                "type": "Int32"
            },
            {                                                                                                    
                "name": "Id",
                "type": "String"
            },
            {
                "name": "DeviceId",
                "type": "String"
            },
            {
                "name": "MessageType",
                "type": "Int32"
            },
            {
                "name": "Longitude",
                "type": "Double"
            },
            {
                "name": "Latitude",
                "type": "Double"
            },
            {
                "name": "Timestamp",
                "type": "Datetime"
            },
            {
                "name": "EventProcessedUtcTime",
                "type": "Datetime"
            },
            {
                "name": "PartitionId",
                "type": "Int32"
            },
            {
                "name": "EventEnqueuedUtcTime",
                "type": "Datetime"
            },
            {
                "name": "FirstName",
                "type": "String"
            },
            {
                "name": "LastName",
                "type": "String"
            },
            {
                "name": "UserName",
                "type": "String"
            },                                                                                                    
            {
                "name": "UType",
                "type": "String"
            },
            {
                "name": "Phone",
                "type": "String"
            },
            {
                "name": "Email",
                "type": "String"
            },
            {
                "name": "Gender",
                "type": "String"
            },
            {
                "name": "MLUserid",
                "type": "String"
            },
            {
                "name": "MLTimestamp",
                "type": "String"
            },
            {
                "name": "ScoredLabels",
                "type": "String"
            },
            {
                "name": "ScoredProb",
                "type": "Double"
            },
            {
                "name": "Race",
                "type": "Int32"
            },
            {
                "name": "CompanyName",
                "type": "String"
            },
            {
                "name": "Rid",
                "type": "String"
            }
        ],
        "published": false,
        "type": "AzureSqlTable",
        "linkedServiceName": "Destination-SQLAzure-f0g",
        "typeProperties": {
            "tableName": "[dbo].[IOTDataMLHistory]"
        },                                                                                                    
        "availability": {
            "frequency": "Day",
            "interval": 1
        },
        "external": false,
        "policy": {}
    }
}

Pipeline : Copy CSV to Azure SQL Database

{
    "name": "CopyPipeline-f0g",
    "properties": {
        "activities": [
            {
                "type": "Copy",
                "typeProperties": {
                    "source": {
                        "type": "AzureDataLakeStoreSource",
                        "recursive": true
                    },
                    "sink": {
                        "type": "SqlSink",
                        "writeBatchSize": 0,
                        "writeBatchTimeout": "00:00:00"
                    },
                    "translator": {
                        "type": "TabularTranslator",
                        "columnMappings": "Column0:UserId,Column1:Age,Column2:Height,Column3:Weight,Column4:HeartRateBPM,Column5:BreathingRate,Column6:Temperature,Column7:Steps,Column8:Velocity,Column9:Altitude,Column10:Ventilization,Column11:Activity,Column12:Cadence,Column13:Speed,Column14:HIB,Column15:HeartRateRedZone,Column16:HeartrateVariability,Column17:Status,Column18:Id,Column19:DeviceId,Column20:MessageType,Column21:Longitude,Column22:Latitude,Column23:Timestamp,Column24:EventProcessedUtcTime,Column25:PartitionId,Column26:EventEnqueuedUtcTime,Column27:FirstName,Column28:LastName,Column29:UserName,Column30:UType,Column31:Phone,Column32:Email,Column33:Gender,Column34:MLUserid,Column35:MLTimestamp,Column36:ScoredLabels,Column37:ScoredProb,Column38:Race,Column39:CompanyName,Column40:Rid"                                                                                                    
                    }
                },
                "inputs": [
                    {
                        "name": "InputDataset-f0g"
                    }
                ],
                "outputs": [
                    {
                        "name": "OutputDataset-f0g"
                    }
                ],
                "policy": {
                    "timeout": "1.00:00:00",
                    "concurrency": 1,
                    "executionPriorityOrder": "NewestFirst",
                    "style": "StartOfInterval",
                    "retry": 3,
                    "longRetry": 0,
                    "longRetryInterval": "00:00:00"
                },
                "scheduler": {
                    "frequency": "Day",
                    "interval": 1
                },
                "name": "Activity-0-Data lake path_ output_->[dbo]_[IOTDataMLHistory]"
            }
        ],
        "start": "2016-12-23T03:32:59.874Z",
        "end": "2099-12-31T05:00:00Z",
        "isPaused": false,
        "hubName": "populateazuresqldbs_hub",
        "pipelineMode": "Scheduled"
    }
}                                                                                                    

Once you have finished creating the new Azure Data Factory job to copy the CSV file data into the Azure SQL databases, you can easily verify the output. Open SQL Server Management Studio or Visual Studio Server Explorer and run a SQL query over the newly populated SQL table. Figure 7-26 illustrates the results of a SQL query against the IOTDataMLHistory SQL table.

Figure 7-26. Verify import Azure SQL database using a SQL query in SQL Server Management Studio

At this point, you have completed all the tasks necessary to implement the Azure architecture for the reference implementation, as illustrated in Figure 7-27.

Figure 7-27. Azure architecture for the reference implementation

A final step in this process for a full production implementation would be to “lock down” the security aspects of this solution by restricting users and user access to the Azure Data Lake Store and Azure SQL databases via the Azure Portal.