Skip to content

module02b.md

Latest commit

 

History

History
513 lines (279 loc) · 15.4 KB

module02b.md

File metadata and controls

513 lines (279 loc) · 15.4 KB
Raw

Module 02B - Incremental Load Fact

< Previous Module - Home - Next Module >

⏱️ Estimated Duration

20 minutes

🤔 Prerequisites

  • Lab environment deployed
  • Module 1A (Linked Service, Integration Datasets)
  • Module 2A complete

📢 Introduction

In this module, we will setup a Synapse Pipeline to incrementally load data from our raw layer (CSV), into our curated layer (Delta Lake) where our fact table resides.

flowchart LR

param((fileName))
param-.->p

ds1[(Data Lake\nraw)]
ds2a[(Data Lake\ncurated)]
ds2b[(Data Lake\ncurated)]

ds1-."01-raw/wwi/orders/$fileName\nCSV".->df1
ds2b-."03-curated/wwi/customers\nDelta Lake".->df7
df6-."03-curated/wwi/orders\nDelta Lake".->ds2a

subgraph p["Pipeline (O2 - pipelineFactIncrementalLoad)"]
a1[Data flow\nincrementalLoadFact]
end

a1-.->df

subgraph df["Data flow (dataFlowFactIncrementalLoad)"]
df1[Source\nrawOrders]
df2[Lookup\nlookupDimCustomer]
df3[Select\nselectFactColumns]
df4[Derived column\ncheckForEarlyFacts]
df5[Alter row\nmarkAsUpsert]
df6[Sink\nsinkOrders]
df7[Source\ndimCustomer]
df8[Filter\nactiveCustomers]
df1-->df2
df2-->df3
df3-->df4
df4-->df5
df5-->df6
df7-->df8
df8-->df2
end
Loading

🎯 Objectives

  • Create a pipeline that will incrementally load data as new files arrive.

Table of Contents

  1. Pipeline (pipelineFactIncrementalLoad)
  2. Data flow (Source - rawOrders)
  3. Data flow (Source - dimCustomer)
  4. Data flow (Filter - activeCustomers)
  5. Data flow (Lookup - lookupDimCustomer)
  6. Data flow (Select - selectFactColumns)
  7. Data flow (Derived column - checkForEarlyFacts)
  8. Data flow (Alter row - markAsUpsert)
  9. Data flow (Sink - sinkOrders)
  10. Pipeline (pipelineFactIncrementalLoad)
  11. Debug Pipeline

1. Pipeline (pipelineFactIncrementalLoad)

In this step, we are going to create a new pipeline O2 - pipelineFactIncrementalLoad that will include a Data flow activity to incrementally load data from raw into our fact table within the curated layer.

  1. Navigate to the Integrate hub

    ALT

  2. Under Pipelines, click on the ellipsis [...] icon to the right of the Orders folder and select New pipeline

    ALT

  3. Rename the pipeline to O2 - pipelineFactIncrementalLoad

    ALT

  4. Under Parameters, click New

    ALT

  5. Set the name of the parameter to fileName

    ALT

  6. Within Activities, search for Data flow, and drag the Data flow activity onto the canvas

    ALT

  7. Rename the activity incrementalLoadFact

    ALT

  8. Switch to the Settings tab

    ALT

  9. Next to the Data flow property, click New

    ALT

2. Data flow (Source - rawOrders)

In this step, we start with a source transformation that will reference a delimited text file (CSV) in the raw layer of our data lake. The data flow will include a file name parameter, this will allow the pipeline to dynamically pass a file name at runtime.

  1. Enable Data flow debug

    ALT

  2. Rename the data flow dataFlowFactIncrementalLoad

    ALT

  3. Under Parameters, click New

    ALT

  4. Rename parameter1 to fileName

    ALT

  5. Within the data flow canvas, click Add Source and select Add source

    ALT

  6. Rename the Output stream name to rawOrders

    ALT

  7. Set the Source type to Inline

    ALT

  8. Set the Inline dataset type to DelimitedText

    ALT

  9. Set the Linked Service to the Synapse Workspace Default Storage

    ALT

  10. Switch to the Source options tab and click the Browse icon

    ALT

  11. Navigate to 01-raw > wwi > orders and click OK

    ALT

  12. Click inside the File name text input and click Add dynamic content

    ALT

  13. Under Expression elements click Parameters, select fileName, and click Save and finish

    ALT

  14. Enable First row as header

    ALT

  15. Switch to the Projection tab and click Import schema

    ALT

  16. Click Import

    ALT

  17. Under Data flow parameters, set the fileName property to an existing CSV file that resides within 01-raw > wwi > orders, and click Save

    • Tip: In a new window, open the Azure Portal, navigate to the storage account, and use the Storage Browser to find an existing file.
    • Note: The string must be wrapped in single quotes.

    ALT

  18. Switch to the Data preview tab and click Refresh

    ALT

3. Data flow (Source - dimCustomer)

In this step, we will add a second source transformation that will reference the existing Customer dimension table (Delta Lake) in the curated layer of our data lake.

  1. Within the data flow canvas, click Add Source and select Add source

    ALT

  2. Rename the Output stream name to dimCustomer

    ALT

  3. Set the Source type to Inline

    ALT

  4. Set the Inline dataset type to Delta

    ALT

  5. Set the Linked Service to the Synapse Workspace Default Storage

    ALT

  6. Switch to the Source options tab and click the Browse icon

    ALT

  7. Navigate to 03-curated > wwi > customers and click OK

    ALT

  8. Set the Compression type to snappy

    ALT

  9. Switch to the Projection tab and click Import schema

    ALT

  10. Click Import

    ALT

  11. Switch to the Data preview tab and click Refresh

    ALT

4. Data flow (Filter - activeCustomers)

In this step, we will filter the Customers dimension table to only include rows that are active. This is a necessary step as we will eventually lookup which customers are being referenced in the incoming orders data by their CustomerID.

  1. Click the [+] icon to the right of dimCustomer, under Row modifier select Filter

    ALT

  2. Rename the Output stream name to activeCustomers

    ALT

  3. Set the Filter on property to IsActive == 1

    ALT

  4. Switch to the Data preview tab and click Refresh

    ALT

5. Data flow (Lookup - lookupDimCustomer)

The lookup transformation references data from a secondary stream, where there is a match, the step will append columns with the columns from the primary stream. In this step, we will lookup customer records from activeCustomers and append matched records with the orders data rawOrders.

  1. Click the [+] icon to the right of rawOrders, under Multiple inputs/outputs select Lookup

    ALT

  2. Rename the Output stream name to lookupDimCustomer

    ALT

  3. Set the Lookup stream to activeCustomers

    ALT

  4. Set the Lookup conditions to CustomerID on both the Left and Right

    ALT

  5. Switch to the Data preview tab and click Refresh

    ALT

6. Data flow (Select - selectFactColumns)

In this step, we will use a select transformation to drop all columns except OrderId, CustomerSK, and Quantity, reorder CustomerSK to the second position, and rename CustomerSK to CustomerKey.

  1. Click the [+] icon to the right of lookupDimCustomer, under Schema modifier select Select

    ALT

  2. Rename the Output stream name to selectFactColumns

    ALT

  3. Under the Input columns, delete all columns except OrderID, CustomerSK, and Quantity

    ALT

  4. On the left hand side of the CustomerSK, click and drag the column to the second position

    ALT

  5. Rename CustomerSK to CustomerKey

    ALT

  6. Switch to the Data preview tab and click Refresh

    ALT

7. Data flow (Derived column - checkForEarlyFacts)

In this step, we are going to update an existing column CustomerKey and set it to 0 if the existing value is null.

  1. Click the [+] icon to the right of selectFactColumns, under Schema modifier select Derived Column

    ALT

  2. Rename the Output stream name to checkForEarlyFacts

    ALT

  3. Under Columns, click the Column drop-down menu and select CustomerKey

    ALT

  4. Copy and paste the code snippet into the Expression

    iif(isNull(CustomerKey),toLong(0),CustomerKey)

    ALT

  5. Switch to the Data preview tab and click Refresh

    ALT

8. Data flow (Alter row - markAsUpsert)

In this step, we are going to mark all rows from the incoming stream with the UPSERT policy.

  1. Click the [+] icon to the right of checkForEarlyFacts, under Row modifier select Alter Row

    ALT

  2. Rename the Output stream name to markAsUpsert

    ALT

  3. Under Alter row conditions, set the condition to Upsert If and the expression as true()

    ALT

  4. Switch to the Data preview tab and click Refresh

    ALT

9. Data flow (Sink - sinkOrders)

In this step, we will write the the results from the incoming stream to the destination Delta Lake table.

  1. Click the [+] icon to the right of markAsUpsert, under Destination select Sink

    ALT

  2. Rename the Output stream name to sinkOrders

    ALT

  3. Set the Sink type to Inline

    ALT

  4. Set the Inline dataset type to Delta

    ALT

  5. Set the Linked Service to the Synapse Workspace Default Storage

    ALT

  6. Switch to the Settings tab and click the Browse icon

    ALT

  7. Navigate to 03-curated > wwi and click OK

    ALT

  8. Within the Folder path property, replace wwi with wwi/orders

    ALT

  9. Set the Compression type to snappy

    ALT

  10. Set the Update method to Allow insert and Allow upsert

    ALT

  11. Set the Key columns to OrderID

    ALT

  12. Switch to the Data preview tab and click Refresh

    ALT

10. Pipeline (pipelineFactIncrementalLoad)

Update the Data Flow activity within the pipeline to pass the pipeline parameter @pipeline().parameters.fileName to the Data Flow parameter fileName.

  1. Navigate back to the pipeline O2 - pipelineFactIncrementalLoad

    ALT

  2. Click to focus on the Data flow activity and switch to the Parameters tab

    ALT

  3. Under Data flow parameters, click inside the fileName Value and select Pipeline expression

    ALT

  4. Copy and paste the code snippet and click OK

    @pipeline().parameters.fileName

    ALT

  5. Click Publish all

    ALT

  6. Click Publish

    ALT

11. Debug Pipeline

To test that our pipeline is working correctly, we will trigger a manual run using the Debug capability.

  1. Click Debug

    ALT

  2. Set the fileName parameter value to the name of an existing CSV file and click OK

    ALT

  3. Periodically click Refresh until the pipeline has succeeded

    ALT

  4. Navigate to the Data hub, browse the data lake folder structure to 03-curated > wwi, right-click the folder orders, and select New SQL Script > Select TOP 100 rows

    ALT

  5. Set the File type to Delta format and click Apply

    ALT

  6. Click Run

    ALT

🎉 Summary

You have successfully setup a pipeline to incrementally load the fact table (Orders) using the Delta Lake format.

✅ Results

Azure Synapse Analytics

  • 1 x Pipeline (O2 - pipelineFactIncrementalLoad)
  • 1 x Data flow (dataFlowFactIncrementalLoad)

Azure Data Lake Storage Gen2

  • 1 x Delta log file (03-curated/wwi/orders/_delta_log)
  • 2 x Parquet files (03-curated/wwi/orders)

Continue >