In this article, we will see how to create an Azure Data Factory and we will copy data from Blob Storage to Cosmos DB using ADF pipelines. We will create the Source and Destination (Sink) datasets in the pipeline and will link these datasets with the Azure subscription. We will publish this pipeline and later, trigger it manually.

Prerequisites

  • Azure Blob Storage Account
  • Cosmos DB Account
Azure Data Factory is a cloud-based data integration service that allows you to create data-driven workflows in the cloud for orchestrating and automating data movement and data transformation. Using Azure Data Factory, you can create and schedule data-driven workflows (called pipelines) that can ingest data from disparate data stores.
Step 1 - Create Azure Data Factory
Log into the Azure Portal.
Select "Create a resource" and choose Analytics -> Data Factory.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Give a valid name to the Azure Data Factory and choose resource group. If you don’t have any existing resource group, please create a new one.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory

Azure Data Factory will be created shortly.

Step 2 - Store Data in Blob Storage

We can upload a sample CSV file to the Blob Storage now.

Go to Storage Account and click “Storage Explorer” (Currently, it is in the preview mode).

Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
By right clicking the Blob Container, you can see the "Create Blob Container" context menu. Just click it.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Please choose a valid container name (it is case-sensitive) and choose Public access level as Container so that we can access this container later from our Azure Data Factory.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Open the container and upload a sample CSV file to the blob container. I will upload an employee data CSV file which contains only 3 records. Please click the “Upload” button to proceed.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory

Step 3 - Create a new Database and Collection in Azure Cosmos DB Account

Open Cosmos DB account and click Data Explorer.

Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Click “New Database” button, give database name, and choose Throughput value. (It is not mandatory, you can simply ignore it).
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
You can add a new collection to this database by right clicking the database and choosing “New Collection” button.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Give a name to the Collection and give partition key also. Partition key is like a Primary key in SQL Server database.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory

Step 4 - Create a Pipeline in Azure Data Factory

We have already created Azure Data Factory. We can create a “Copy Data” pipeline now. Please open Azure Data Factory and click “Author and Monitor” button

Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
It will open the ADF dashboard. Choose “Create Pipeline” option.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
We can create a new activity now. In the filer box, please type “Copy” it will show the “Copy Data” option under Move & Transform tab. You can drag this activity to the work area as I did.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
We can rename the Activity in “General” tab. I have given a small description also.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory

In the Source tab, you can select the source dataset. Please click “New” Button. It will list all the data sources available in ADF. Currently, Microsoft supports more than 70 data sources.

As our source dataset is Blob storage, please choose it and click the “Finish” Button.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Choose the “Connection” tab and click “New” button to create a new linked service for source dataset.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
You can choose your Azure subscription and choose the already created Storage account name and click the “Finish” button.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
We can choose container name in Connection by browsing it and choose the container name from blob storage.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory

You can ignore the file name. It will automatically pick the file from the container.

In our CSV, the first row contains the column name. Select “Column name in first row” option.

Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory

You can click the pipeline and choose the Sink tab. It is for Destination dataset.

Please click the “New” button to choose Destination dataset and select Cosmos DB as destination data source and click “Finish” button.

Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Choose Connection tab and click “New” button to create new linked service for Cosmos DB.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Choose your Azure subscription from the dropdown list and select Cosmos DB account name and database name from the list and click “Finish” button.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
You can choose the Collection name from the list. (We have already created the collection in Cosmos DB account)
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
We have created Source Dataset and Sink Dataset in our pipeline. We can validate the pipeline and datasets before publishing it.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
We can see the validation errors if anything occurred.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Our validation was successful. We can now publish ADF.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory

It will take some time to publish all the changes.

After a successful publish, we can Trigger the pipeline.

Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory

Click “Trigger” Button and choose “Trigger Now” It will open a window and choose “Finish” button.

We will be notified with a message that the pipeline succeeded.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory
Our data integration is completed now. We can open the Cosmos DB to check the copied data from Blob Storage. You can see that there are three records (documents) available in Cosmos DB. As I mentioned earlier my CSV file contains 3 records.
Copy Data From Blob Storage To Cosmos DB Using Azure Data Factory

You can download the ARM (Azure Resource Manager) template for this ADF for future use. ARM template contains the pipeline and dataset details.

Normally there are two ARM templates available for each ADF.

“arm_template.json” and “arm_template_parameters.json”
arm_template.json
  1. {
  2. "$schema": "http://schema.management.azure.com/schemas/2015-01-01/deploymentTemplate.json#",
  3. "contentVersion": "1.0.0.0",
  4. "parameters": {
  5. "factoryName": {
  6. "type": "string",
  7. "metadata": "Data Factory Name",
  8. "defaultValue": "sarathadf1"
  9. },
  10. "AzureBlobStorage1_connectionString": {
  11. "type": "secureString",
  12. "metadata": "Secure string for 'connectionString' of 'AzureBlobStorage1'"
  13. },
  14. "CosmosDb1_connectionString": {
  15. "type": "secureString",
  16. "metadata": "Secure string for 'connectionString' of 'CosmosDb1'"
  17. },
  18. "AzureBlob1_properties_typeProperties_folderPath": {
  19. "type": "string",
  20. "defaultValue": "sarathcontainer"
  21. }
  22. },
  23. "variables": {
  24. "factoryId": "[concat('Microsoft.DataFactory/factories/', parameters('factoryName'))]"
  25. },
  26. "resources": [
  27. {
  28. "name": "[concat(parameters('factoryName'), '/AzureBlobStorage1')]",
  29. "type": "Microsoft.DataFactory/factories/linkedServices",
  30. "apiVersion": "2018-06-01",
  31. "properties": {
  32. "annotations": [],
  33. "type": "AzureBlobStorage",
  34. "typeProperties": {
  35. "connectionString": "[parameters('AzureBlobStorage1_connectionString')]"
  36. }
  37. },
  38. "dependsOn": []
  39. },
  40. {
  41. "name": "[concat(parameters('factoryName'), '/pipeline1')]",
  42. "type": "Microsoft.DataFactory/factories/pipelines",
  43. "apiVersion": "2018-06-01",
  44. "properties": {
  45. "activities": [
  46. {
  47. "name": "BlobCopyToCosmosDB",
  48. "description": "Copy data from Blob Storage to Cosmos DB Account.",
  49. "type": "Copy",
  50. "dependsOn": [],
  51. "policy": {
  52. "timeout": "7.00:00:00",
  53. "retry": 0,
  54. "retryIntervalInSeconds": 30,
  55. "secureOutput": false,
  56. "secureInput": false
  57. },
  58. "userProperties": [],
  59. "typeProperties": {
  60. "source": {
  61. "type": "BlobSource",
  62. "recursive": true
  63. },
  64. "sink": {
  65. "type": "DocumentDbCollectionSink",
  66. "nestingSeparator": ".",
  67. "writeBatchSize": 10000,
  68. "writeBehavior": "insert"
  69. },
  70. "enableStaging": false,
  71. "dataIntegrationUnits": 0
  72. },
  73. "inputs": [
  74. {
  75. "referenceName": "AzureBlob1",
  76. "type": "DatasetReference",
  77. "parameters": {}
  78. }
  79. ],
  80. "outputs": [
  81. {
  82. "referenceName": "DocumentDbCollection1",
  83. "type": "DatasetReference",
  84. "parameters": {}
  85. }
  86. ]
  87. }
  88. ],
  89. "annotations": []
  90. },
  91. "dependsOn": [
  92. "[concat(variables('factoryId'), '/datasets/AzureBlob1')]",
  93. "[concat(variables('factoryId'), '/datasets/DocumentDbCollection1')]"
  94. ]
  95. },
  96. {
  97. "name": "[concat(parameters('factoryName'), '/AzureBlob1')]",
  98. "type": "Microsoft.DataFactory/factories/datasets",
  99. "apiVersion": "2018-06-01",
  100. "properties": {
  101. "linkedServiceName": {
  102. "referenceName": "AzureBlobStorage1",
  103. "type": "LinkedServiceReference"
  104. },
  105. "annotations": [],
  106. "type": "AzureBlob",
  107. "typeProperties": {
  108. "format": {
  109. "type": "TextFormat",
  110. "columnDelimiter": ",",
  111. "nullValue": "\\N",
  112. "treatEmptyAsNull": true,
  113. "skipLineCount": 0,
  114. "firstRowAsHeader": true
  115. },
  116. "folderPath": "[parameters('AzureBlob1_properties_typeProperties_folderPath')]"
  117. }
  118. },
  119. "dependsOn": [
  120. "[concat(variables('factoryId'), '/linkedServices/AzureBlobStorage1')]"
  121. ]
  122. },
  123. {
  124. "name": "[concat(parameters('factoryName'), '/DocumentDbCollection1')]",
  125. "type": "Microsoft.DataFactory/factories/datasets",
  126. "apiVersion": "2018-06-01",
  127. "properties": {
  128. "linkedServiceName": {
  129. "referenceName": "CosmosDb1",
  130. "type": "LinkedServiceReference"
  131. },
  132. "annotations": [],
  133. "type": "DocumentDbCollection",
  134. "typeProperties": {
  135. "collectionName": "employee"
  136. }
  137. },
  138. "dependsOn": [
  139. "[concat(variables('factoryId'), '/linkedServices/CosmosDb1')]"
  140. ]
  141. },
  142. {
  143. "name": "[concat(parameters('factoryName'), '/CosmosDb1')]",
  144. "type": "Microsoft.DataFactory/factories/linkedServices",
  145. "apiVersion": "2018-06-01",
  146. "properties": {
  147. "annotations": [],
  148. "type": "CosmosDb",
  149. "typeProperties": {
  150. "connectionString": "[parameters('CosmosDb1_connectionString')]"
  151. }
  152. },
  153. "dependsOn": []
  154. }
  155. ]
  156. }

arm_template_parameters.json

  1. {
  2. "$schema": "https://schema.management.azure.com/schemas/2015-01-01/deploymentParameters.json#",
  3. "contentVersion": "1.0.0.0",
  4. "parameters": {
  5. "factoryName": {
  6. "value": "sarathadf1"
  7. },
  8. "AzureBlobStorage1_connectionString": {
  9. "value": ""
  10. },
  11. "CosmosDb1_connectionString": {
  12. "value": ""
  13. },
  14. "AzureBlob1_properties_typeProperties_folderPath": {
  15. "value": "sarathcontainer"
  16. }
  17. }
  18. }

First json file contains all the pipeline and dataset information and second json file contains the details about parameters. You can give the storage account and other connection details in this file. We can import these ARM templates in future and save the time.

In this article, we have created an Azure Data Factory and we have uploaded one simple CSV file to Blob Storage and we have created one Database and empty Collection in Cosmos DB also. We have created a pipeline with two datasets in Data Factory. One for Azure Blob storage account and other for Cosmos DB account. We have linked both datasets with the Azure subscription. Later we published the pipeline and dataset. We have seen how to trigger pipeline on-demand also. We have seen all the data successfully copied from Blob storage to Cosmos DB.