dennis/Azure-Data-Integration-Pipeline-Training
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'main' of https://github.com/mrpaulandrewltd/Azure-Data-Integration-Pipeline-Training

Browse Source
This commit is contained in:
Paul Andrew 2022-10-27 11:23:57 +01:00
commit 820694908f
20 changed files with 1485 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
Code/DataFactory/dataflow/MappingOrderAggregationWithParam.json Normal file
View File

@ -0,0 +1,45 @@
{
"name": "MappingOrderAggregationWithParam",
"properties": {
"type": "MappingDataFlow",
"typeProperties": {
"sources": [
{
"dataset": {
"referenceName": "LakeFileOrderHeaderParquet",
"type": "DatasetReference"
},
"name": "OrderHeader"
},
{
"dataset": {
"referenceName": "LakeFileOrderDetailLinesParquet",
"type": "DatasetReference"
},
"name": "OrderLineDetails"
}
],
"sinks": [
{
"dataset": {
"referenceName": "TableOrderSummary",
"type": "DatasetReference"
},
"name": "OrderSummary"
}
],
"transformations": [
{
"name": "JoinHeaderToLineDetails"
},
{
"name": "OrderLineCount"
},
{
"name": "AddAuditColum"
}
],
"script": "parameters{\n\tAuditColumn as string\n}\nsource(output(\n\t\tSalesOrderID as integer,\n\t\tRevisionNumber as integer,\n\t\tOrderDate as timestamp,\n\t\tDueDate as timestamp,\n\t\tShipDate as timestamp,\n\t\tStatus as integer,\n\t\tOnlineOrderFlag as boolean,\n\t\tSalesOrderNumber as string,\n\t\tPurchaseOrderNumber as string,\n\t\tAccountNumber as string,\n\t\tCustomerID as integer,\n\t\tShipToAddressID as integer,\n\t\tBillToAddressID as integer,\n\t\tShipMethod as string,\n\t\tCreditCardApprovalCode as string,\n\t\tSubTotal as decimal(19,4),\n\t\tTaxAmt as decimal(19,4),\n\t\tFreight as decimal(19,4),\n\t\tTotalDue as decimal(19,4),\n\t\tComment as string,\n\t\trowguid as string,\n\t\tModifiedDate as timestamp\n\t),\n\tallowSchemaDrift: false,\n\tvalidateSchema: false,\n\tignoreNoFilesFound: false,\n\tformat: 'parquet',\n\tpartitionBy('hash', 1)) ~> OrderHeader\nsource(output(\n\t\tSalesOrderID as integer,\n\t\tSalesOrderDetailID as integer,\n\t\tOrderQty as integer,\n\t\tProductID as integer,\n\t\tUnitPrice as decimal(19,4),\n\t\tUnitPriceDiscount as decimal(19,4),\n\t\tLineTotal as decimal(38,6),\n\t\trowguid as string,\n\t\tModifiedDate as timestamp\n\t),\n\tallowSchemaDrift: false,\n\tvalidateSchema: false,\n\tignoreNoFilesFound: false,\n\tformat: 'parquet',\n\tpartitionBy('hash', 1)) ~> OrderLineDetails\nOrderHeader, OrderLineDetails join(OrderHeader@SalesOrderID == OrderLineDetails@SalesOrderID,\n\tjoinType:'inner',\n\tmatchType:'exact',\n\tignoreSpaces: false,\n\tpartitionBy('hash', 1),\n\tbroadcast: 'both')~> JoinHeaderToLineDetails\nJoinHeaderToLineDetails aggregate(groupBy(SalesOrderNumber),\n\tRecordCount = count(SalesOrderDetailID),\n\tpartitionBy('roundRobin', 4)) ~> OrderLineCount\nOrderLineCount derive(AuditValue = $AuditColumn) ~> AddAuditColum\nAddAuditColum sink(allowSchemaDrift: false,\n\tvalidateSchema: false,\n\tinput(\n\t\tSalesOrderNumber as string,\n\t\tRecordCount as integer\n\t),\n\tdeletable:false,\n\tinsertable:true,\n\tupdateable:false,\n\tupsertable:false,\n\ttruncate:true,\n\tformat: 'table',\n\tskipDuplicateMapInputs: true,\n\tskipDuplicateMapOutputs: true,\n\terrorHandlingOption: 'stopOnFirstError',\n\tmapColumn(\n\t\tSalesOrderNumber,\n\t\tRecordCount\n\t),\n\tpartitionBy('roundRobin', 4)) ~> OrderSummary"
}
}
}

2
Code/DataFactory/dataset/ADLS_BIN_AWProduct.json
View File

@ -6,7 +6,7 @@
"type": "LinkedServiceReference"
},
"folder": {
"name": "Labs"
"name": "Labs1"
},
"annotations": [],
"type": "Binary",

2
Code/DataFactory/dataset/ADLS_BIN_AdventureWorks.json
View File

@ -11,7 +11,7 @@
}
},
"folder": {
"name": "Labs"
"name": "Labs1"
},
"annotations": [],
"type": "Binary",

32
Code/DataFactory/dataset/ADLS_PQT_AdventureWorks.json Normal file
View File

@ -0,0 +1,32 @@
{
"name": "ADLS_PQT_AdventureWorks",
"properties": {
"linkedServiceName": {
"referenceName": "ADLS_saintegrationpipelines",
"type": "LinkedServiceReference"
},
"parameters": {
"EntityName": {
"type": "string"
}
},
"folder": {
"name": "Labs2"
},
"annotations": [],
"type": "Parquet",
"typeProperties": {
"location": {
"type": "AzureBlobFSLocation",
"fileName": {
"value": "@{dataset().EntityName}.parquet",
"type": "Expression"
},
"folderPath": "Raw",
"fileSystem": "lakeroot"
},
"compressionCodec": "snappy"
},
"schema": []
}
}

2
Code/DataFactory/dataset/ADLS_TSV_AdventureWorks.json
View File

@ -11,7 +11,7 @@
}
},
"folder": {
"name": "Labs"
"name": "Labs1"
},
"annotations": [],
"type": "DelimitedText",

2
Code/DataFactory/dataset/HTTP_BIN_AWProduct.json
View File

@ -6,7 +6,7 @@
"type": "LinkedServiceReference"
},
"folder": {
"name": "Labs"
"name": "Labs1"
},
"annotations": [],
"type": "Binary",

2
Code/DataFactory/dataset/HTTP_BIN_AdventureWorks.json
View File

@ -11,7 +11,7 @@
}
},
"folder": {
"name": "Labs"
"name": "Labs1"
},
"annotations": [],
"type": "Binary",

32
Code/DataFactory/dataset/HTTP_TSV_AdventureWorks.json Normal file
View File

@ -0,0 +1,32 @@
{
"name": "HTTP_TSV_AdventureWorks",
"properties": {
"linkedServiceName": {
"referenceName": "HTTP_AWGitHub",
"type": "LinkedServiceReference"
},
"parameters": {
"EntityName": {
"type": "string"
}
},
"folder": {
"name": "Labs2"
},
"annotations": [],
"type": "DelimitedText",
"typeProperties": {
"location": {
"type": "HttpServerLocation",
"relativeUrl": {
"value": "@{dataset().EntityName}.csv",
"type": "Expression"
}
},
"columnDelimiter": "\t",
"escapeChar": "\\",
"quoteChar": "\""
},
"schema": []
}
}

3
Code/DataFactory/factory/TrainingFactoryDev.json
View File

@ -10,6 +10,9 @@
"type": "string",
"value": "trainingkeys01"
}
},
"globalConfigurations": {
"PipelineBillingEnabled": "true"
}
},
"location": "uksouth",

13
Code/DataFactory/linkedService/EmailSenderFunction.json Normal file
View File

@ -0,0 +1,13 @@
{
"name": "EmailSenderFunction",
"type": "Microsoft.DataFactory/factories/linkedservices",
"properties": {
"annotations": [],
"type": "AzureFunction",
"typeProperties": {
"functionAppUrl": "https://frameworksupportfunctions.azurewebsites.net",
"authentication": "Anonymous",
"encryptedCredential": "ew0KICAiVmVyc2lvbiI6ICIyMDE3LTExLTMwIiwNCiAgIlByb3RlY3Rpb25Nb2RlIjogIktleSIsDQogICJTZWNyZXRDb250ZW50VHlwZSI6ICJQbGFpbnRleHQiLA0KICAiQ3JlZGVudGlhbElkIjogIkRBVEFGQUNUT1JZQEMzMkUyRjlELTM1NzUtNDQ5Qy1BMEQyLTI0Qjg3Rjk5RUFFRV9jZGYyMmI4Ni0zNTk2LTRkNzktYjdmOC1kNGNiMjUyNDZmYjUiDQp9"
}
}
}

2
Code/DataFactory/pipeline/BuildDimProduct.json
View File

@ -39,7 +39,7 @@
}
],
"folder": {
"name": "Labs"
"name": "Labs1"
},
"annotations": []
}

2
Code/DataFactory/pipeline/CopyAWEntities.json
View File

@ -48,7 +48,7 @@
}
},
"folder": {
"name": "Labs"
"name": "Labs1"
},
"annotations": []
}

2
Code/DataFactory/pipeline/CopyAWEntity.json
View File

@ -65,7 +65,7 @@
}
},
"folder": {
"name": "Labs"
"name": "Labs1"
},
"annotations": []
}

2
Code/DataFactory/pipeline/CopyAWProduct.json
View File

@ -48,7 +48,7 @@
}
],
"folder": {
"name": "Labs"
"name": "Labs1"
},
"annotations": []
}

170
Code/DataFactory/pipeline/CopyAWUsingMetadata.json Normal file
View File

@ -0,0 +1,170 @@
{
"name": "CopyAWUsingMetadata",
"properties": {
"activities": [
{
"name": "Get catalog",
"type": "WebActivity",
"dependsOn": [],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"url": "https://raw.githubusercontent.com/mrpaulandrewltd/Azure-Data-Integration-Pipeline-Training/main/Labs/TableCatalog.json",
"method": "GET"
}
},
{
"name": "Cache catalog as JSON array",
"type": "SetVariable",
"dependsOn": [
{
"activity": "Get catalog",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"variableName": "Catalog",
"value": {
"value": "@json(activity('Get catalog').output.Response)",
"type": "Expression"
}
}
},
{
"name": "ForEach AW entity",
"type": "ForEach",
"dependsOn": [
{
"activity": "Cache catalog as JSON array",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"items": {
"value": "@variables('Catalog')",
"type": "Expression"
},
"activities": [
{
"name": "Copy AW entity",
"type": "Copy",
"dependsOn": [],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"source": {
"type": "DelimitedTextSource",
"storeSettings": {
"type": "HttpReadSettings",
"requestMethod": "GET"
},
"formatSettings": {
"type": "DelimitedTextReadSettings"
}
},
"sink": {
"type": "ParquetSink",
"storeSettings": {
"type": "AzureBlobFSWriteSettings"
},
"formatSettings": {
"type": "ParquetWriteSettings"
}
},
"enableStaging": false,
"translator": {
"value": "@item().translator",
"type": "Expression"
}
},
"inputs": [
{
"referenceName": "HTTP_TSV_AdventureWorks",
"type": "DatasetReference",
"parameters": {
"EntityName": {
"value": "@item().EntityName",
"type": "Expression"
}
}
}
],
"outputs": [
{
"referenceName": "ADLS_PQT_AdventureWorks",
"type": "DatasetReference",
"parameters": {
"EntityName": {
"value": "@item().EntityName",
"type": "Expression"
}
}
}
]
},
{
"name": "Notify error",
"type": "AzureFunctionActivity",
"dependsOn": [
{
"activity": "Copy AW entity",
"dependencyConditions": [
"Failed"
]
}
],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"functionName": "SendEmail",
"method": "POST",
"body": {
"value": "{\n \"emailRecipients\": \"richard.swinbank@gmail.com\",\n \"emailSubject\": \"Error copying @{item().EntityName}\",\n \"emailBody\": \"@{replace(activity('Copy AW entity').output.errors[0].Message,'\"','\\\"')}\"\n}",
"type": "Expression"
}
},
"linkedServiceName": {
"referenceName": "EmailSenderFunction",
"type": "LinkedServiceReference"
}
}
]
}
}
],
"variables": {
"Catalog": {
"type": "Array",
"defaultValue": []
}
},
"folder": {
"name": "Labs2"
},
"annotations": []
}
}

29
Code/DataFactory/pipeline/Custom Activity.json
View File

@ -30,35 +30,6 @@
"referenceName": "BatchForTraining01",
"type": "LinkedServiceReference"
}
},
{
"name": "HelloWorld - Again",
"type": "Custom",
"dependsOn": [],
"policy": {
"timeout": "7.00:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"command": "CustomActivity.exe",
"resourceLinkedService": {
"referenceName": "TrainingStore01",
"type": "LinkedServiceReference"
},
"folderPath": "customactivities",
"referenceObjects": {
"linkedServices": [],
"datasets": []
}
},
"linkedServiceName": {
"referenceName": "BatchForTraining01",
"type": "LinkedServiceReference"
}
}
],
"folder": {

51
Code/DataFactory/pipeline/EmailSender.json Normal file
View File

@ -0,0 +1,51 @@
{
"name": "EmailSender",
"properties": {
"activities": [
{
"name": "Send Email",
"type": "AzureFunctionActivity",
"dependsOn": [],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"functionName": "SendEmail",
"method": "POST",
"body": {
"value": "@concat('\n{\n\"emailRecipients\": \"',pipeline().parameters.Recipients,'\",\n\"emailSubject\": \"',pipeline().parameters.Subject,'\",\n\"emailBody\": \"',pipeline().parameters.Body,'\",\n\"emailImportance\": \"',pipeline().parameters.Importance,'\"\n}')",
"type": "Expression"
}
},
"linkedServiceName": {
"referenceName": "EmailSenderFunction",
"type": "LinkedServiceReference"
}
}
],
"parameters": {
"Recipients": {
"type": "string"
},
"Subject": {
"type": "string"
},
"Body": {
"type": "string"
},
"Importance": {
"type": "string",
"defaultValue": "High"
}
},
"folder": {
"name": "Demo Pipelines/Misc"
},
"annotations": []
}
}

48
Code/DataFactory/pipeline/Order Summary with Mapping With Param.json Normal file
View File

@ -0,0 +1,48 @@
{
"name": "Order Summary with Mapping With Param",
"properties": {
"activities": [
{
"name": "Mapping Order Aggregation",
"type": "ExecuteDataFlow",
"dependsOn": [],
"policy": {
"timeout": "7.00:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"dataflow": {
"referenceName": "MappingOrderAggregationWithParam",
"type": "DataFlowReference",
"parameters": {
"AuditColumn": {
"value": "'@{pipeline().parameters.AuditColumnValue}'",
"type": "Expression"
}
}
},
"compute": {
"coreCount": 8,
"computeType": "General"
},
"traceLevel": "Fine"
}
}
],
"parameters": {
"AuditColumnValue": {
"type": "string"
}
},
"folder": {
"name": "Demo Pipelines/Data Flows"
},
"annotations": [],
"lastPublishTime": "2020-09-02T14:13:15Z"
},
"type": "Microsoft.DataFactory/factories/pipelines"
}

BIN
Labs/Lab 7 - Mini-project.pdf Normal file
View File

Binary file not shown.

1082
Labs/TableCatalog.json Normal file
View File

File diff suppressed because it is too large Load Diff