{"id":9391,"date":"2022-01-31T08:00:52","date_gmt":"2022-01-31T16:00:52","guid":{"rendered":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/?p=9391"},"modified":"2022-05-26T15:37:31","modified_gmt":"2022-05-26T22:37:31","slug":"combining-microsoft-graph-data-connect-data-sets-in-azure-synapse-analytics","status":"publish","type":"post","link":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/combining-microsoft-graph-data-connect-data-sets-in-azure-synapse-analytics\/","title":{"rendered":"Combining Microsoft Graph Data Connect data sets in Azure Synapse Analytics"},"content":{"rendered":"

This article covers the process of combining two data sets extracted via an Azure Synapse pipeline using Microsoft Graph Data Connect (MGDC). The steps to deploy the baseline Azure Synapse Analytics workspace to follow this demo are described in my blog here<\/a>. For users who are not familiar with Azure Synapse analytics, it is a solution that provides a full Extract\/Transform\/Load (ETL) stack for your data.<\/p>\n

We will demo two different approaches to combining data sets via Azure Synapse Analytics pipelines. The first approach will use Synapse Spark Notebook with PySpark scripts, while the second one will use the no-code Data flow approach instead. In both cases we will demo how to effectively combine the Microsoft 365 Messages data set, which contains emails, with the Microsoft 365 Users data set. We will be joining the two data sets on the pUser property which is common to both sets and identifies the record of specific users.<\/p>\n

Option 1 – Using Synapse Spark Notebook<\/h2>\n

To get started, we will need to create a new Synapse pipeline. To do so, navigate to your Azure Synapse workspace, and open the Synapse Studio. From the main workbench, click on the Integrate<\/strong> button in the left navigation bar.<\/p>\n

\"Azure<\/a><\/p>\n

Figure 1 – Azure Synapse Analytics’ Integrate feature<\/em><\/span><\/p>\n

Beside the Integrate<\/strong> header, click on the +<\/strong> button and select Pipeline<\/strong> from the drop-down menu.<\/p>\n

\"-<\/a><\/p>\n

Figure 2 – Creating a new Azure Synapse Analytics pipeline<\/span><\/em><\/p>\n

We now need to add two Copy Data<\/strong> activities to our pipeline: one that will copy the Users (BasicDataSet_v0.User_v1) data set and one that will copy the Messages (BasicDataSet_v0.Message_v1) one. In our case, we have defined both activities so that they will return all fields from both data sets. For more details on how to configure Microsoft Graph Data Connect via the Copy Data activity, you can refer to the following article: Build your first Microsoft Graph Data Connect application<\/a>.<\/p>\n

For our demos, the extracted data will be copied to an Azure Data Lake Storage Gen 2 location. Also, it is important to note that the activity to extract information about messages will have a dynamic filter on the sentDateTime so that it only extracts information from the past 24 hours. Figure 3 below shows the new pipeline with the 2 Copy Data activities on the workbench.<\/p>\n

\"Azure<\/a><\/p>\n

Figure 3 – Azure Synapse Analytics pipeline with MGDC copy data activities<\/span><\/em><\/p>\n

Every time our pipeline executes, it will retrieve all emails that were sent in the past 24 hours (based on the dynamic filter mentioned previously) and will retrieve information about all users and have it stored in our storage account as binary files with JSON rows. We\u2019ve configured our pipeline so that each activity copies its extracted data in a folder named accordingly (e.g., messages and users folders). For our scenario, we will schedule our pipeline to automatically get triggered every 24 hours, which means that every day it will generate a new binary file containing all emails from the past day. The issue, however, is that the same thing will happen with the users data extract, which will result in duplicate users being listed across files. We want to make sure that every time we run our pipeline we start with a fresh and most recent list of all users in our environment. To enforce this, we will add a new Delete<\/strong> activity (under the General category) and ensure all files are deleted from our users<\/strong> folder as a prerequisites to extracting the users data set.<\/p>\n

\"Adding<\/a><\/p>\n

Figure 4 – Adding a new Delete activity to our pipeline<\/span><\/em><\/p>\n

We will configure the delete activity to delete every file under the specified location as shown in Figure 5 below.<\/p>\n

\"Delete<\/a><\/p>\n

Figure 5 – Delete all files in the specified location<\/em><\/span><\/p>\n

Under the Logging settings<\/strong> tab, we will simply uncheck the Enable logging<\/strong> checkbox.<\/p>\n

\"Disabling<\/a><\/p>\n

Figure 6 – Disabling logging on the delete activity<\/em><\/span><\/p>\n

We are now ready to tackle the bulk of pipeline\u2019s logic. From the Activities<\/strong> list, under the Synapse<\/strong> category, drag and drop a new Notebook<\/strong> activity onto the workbench, and make the two copy data activity prerequisites (see Figure 7).<\/p>\n

\"Add<\/a><\/p>\n

Figure 7 – Add a new Synapse Runbook to the pipeline<\/em><\/span><\/p>\n

From the workbench, select the newly added Notebook activity and in the property panel at the bottom, select the Settings<\/strong> tab. From there, click on the + New<\/strong> button to create a new notebook.<\/p>\n

\"Creating<\/a><\/p>\n

Figure 8 – Creating a new Azure Synapse Notebook<\/span><\/em><\/p>\n

This will launch the Synapse Notebook editor. In the top menu bar, click on Attach to<\/strong> and select your Apache Spark pool from the list.<\/p>\n

\"Selecting<\/a><\/p>\n

Figure 9 – Selecting Apache Spark Pool to run an Azure Synapse Notebook<\/em><\/span><\/p>\n

In the script editor, copy and paste the following PySpark script, making sure to update the value of the filePathMessages<\/strong> and filePathUsers<\/strong> variables with the path to the location where your data is being extracted.<\/p>\n

from pyspark.sql import functions as f\r\nfilePathMessages = \"wasbs:\/\/<messages folder location>\"\r\nfilePathUsers = \"wasbs:\/\/<users folder location>\"\r\nstorageAccountName = \"<storage account name>\"\r\nstorageAccountKey = \"<storage account key>\"\r\nfileStorageUrl = \"fs.azure.account.key.\" + storageAccountName + \r\n\".blob.core.windows.net\" \r\n\r\nspark.conf.set(fileStorageUrl, storageAccountKey)\r\n\r\nmessageDataframe = spark.read.json(filePathMessages).withColumn(\"sender_name\", \r\nf.col(\"sender.emailAddress.name\")).withColumn(\"sender_address\", \r\nf.col(\"sender.emailAddress.address\")).withColumn(\"from_name\", \r\nf.col(\"from.emailAddress.name\")).withColumn(\"from_address\", \r\nf.col(\"from.emailAddress.address\")).withColumn(\"to_name\", \r\nf.col(\"toRecipients.emailAddress.name\")).withColumn(\"to_address\", \r\nf.col(\"toRecipients.emailAddress.address\"))\r\ntry:\r\n      # There is a chance that all entries in the ccRecipients column be empty \r\nwhich will detect the schema as being a string array.<\/em>\r\n      # If it's the case, simply add the derived name and address columns as null<\/em>\r\n      data_type = dict(messageDataframe.dtypes)['ccRecipients']\r\n      if data_type == \"array<string>\":\r\n            messageDataframe = messageDataframe.withColumn(\"cc_name\", f.lit(\"\"))\r\n            messageDataframe = messageDataframe.withColumn(\"cc_address\", \r\nf.lit(\"\"))\r\n      else:\r\n            messageDataframe = messageDataframe.withColumn(\"cc_name\", \r\nf.col(\"ccRecipients.emailAddress.name\"))\r\n            messageDataframe = messageDataframe.withColumn(\"cc_address\", \r\nf.col(\"ccRecipients.emailAddress.address\"))\r\n\r\n      # There is a chance that all entries in the bccRecipients column be empty \r\nwhich will detect the schema as being a string array.\r\n      # If it's the case, simply add the derived name and address columns as null<\/em>\r\n      data_type = dict(messageDataframe.dtypes)['bccRecipients']\r\n      if data_type == \"array<string>\":\r\n            messageDataframe = messageDataframe.withColumn(\"bcc_name\", f.lit(\"\"))\r\n            messageDataframe = messageDataframe.withColumn(\"bcc_address\", f.lit(\"\"))\r\n      else:\r\n            messageDataframe = messageDataframe.withColumn(\"bcc_name\", \r\nf.col(\"bccRecipients.emailAddress.name\"))\r\n            messageDataframe = messageDataframe.withColumn(\"bcc_address\", \r\nf.col(\"bccRecipients.emailAddress.address\"))\r\nfinally:\r\n      messageDataframe = messageDataframe.drop(\"from\", \"sender\", \"toRecipients\", \r\n\"ccRecipients\", \"bccRecipients\")\r\n\r\nusersDataframe = spark.read.json(filePathUsers).withColumn(\"skuIds\", \r\nf.col(\"assignedLicenses.skuId\")).withColumn(\"assignedPlanDates\", \r\nf.col(\"assignedPlans.assignedDateTime\")).withColumn(\"assignedPlanServices\", \r\nf.col(\"assignedPlans.service\")).withColumn(\"puser2\", \r\nf.col(\"puser\")).drop(\"createddatetime\",\"datarow\",\"id\",\"padditionalInfo\",\"ptenant\", \r\n\"rowInformation\", \"userrow\", \"puser\", \"pagerow\")\r\nresults = messageDataframe.join(usersDataframe, messageDataframe.puser == \r\nusersDataframe.puser2,\"inner\").drop(\"puser2\")\r\n\r\nresults.write.mode(\"overwrite\").saveAsTable(\"Results\")<\/pre>\n
 <\/p>\n
The script will flatten the JSON\u2019s hierarchy and rename some fields that are duplicated between the two data sets. It will then merge the two data sets and store them in a new table named Results<\/strong> in our default Synapse Lake database. We are now ready to publish our pipeline and then trigger its execution.<\/p>\n
\"Manually<\/a><\/p>\n
Figure 10 – Manually trigger an Azure Synapse pipeline<\/em><\/span><\/p>\n
Executing the pipeline can take several minutes to an hour to complete its execution once approvers have approved the request, depending on the size of the data you are trying to export. Make sure you do not have any pending approvals for your pipeline so that the pipeline can successfully execute. Once its execution succeeds you can go and browse the default Synapse Lake database to visualize the combined data. To do so, click on the Data<\/strong> button in the left navigation (database icon), expand the Lake database<\/strong> menu, then the default<\/strong> database. Expand the table<\/strong> folder and click on the ellipses beside the newly created Results<\/strong> database. From the flyout menu, select New SQL Script<\/strong> and then Select TOP 100 rows<\/strong>.<\/p>\n
\"Select<\/a><\/p>\n
Figure 11 – Select Top 100 Rows in Lake Database<\/em><\/span><\/p>\n
Next, execute the query by clicking on the Run<\/strong> button at the top of the query editor.<\/p>\n
\"Executing<\/a><\/p>\n
Figure 12 – Executing a SQL query in Azure Synapse Analytics<\/span><\/em><\/p>\n
This will take a few seconds to execute. Once it succeeds, you should be able to review the merged data in your lake database.<\/p>\n
\"Image<\/a><\/p>\n
Figure 13 – Reviewing Merge Data in Query Editor<\/span><\/em><\/p>\n
Option 2 \u2013 Using the No Code Data Flow Activity<\/h2>\n
If you are less familiar with Spark coding or if you simply want a no code solution for your pipeline, you can use the Data Flow<\/strong> activity inside of your pipeline to perform transformation on your extracted data. For this demo, we will clone the pipeline we created with Option 1 above, with the one difference that we will remove the Notebook<\/strong> activity and replace it with a Data flow <\/strong>activity instead. The figure below shows the overview of what the pipeline will look like.<\/p>\n
\"Replacing<\/a><\/p>\n
Figure 14 – Replacing the Notebook activity by a Data flow activity<\/span><\/em><\/p>\n
To edit the flow\u2019s logic, select the new Data flow activity from the workbench and in the property panel at the bottom navigate to the Settings<\/strong> tab. From there, click on the + New<\/strong> button to create a new flow.<\/p>\n
\"Creating<\/a><\/p>\n
Figure 15 – Creating a new data flow<\/span><\/em><\/p>\n
This will launch the data flow editor. The first thing we want to do is load the data we\u2019ve exported as binary file and flatten their hierarchies. Let us start with the Messages data set. From the workbench, click on the arrow beside the Add Source<\/strong> shape and select Add Source<\/strong> from the flyout menu.<\/p>\n
\"Adding<\/a><\/p>\n
Figure 16 – Adding a new data source to our flow<\/em><\/span><\/p>\n
From the property panel at the bottom, click on the + New<\/strong> button beside the Dataset<\/strong> field.<\/p>\n
\"Creating<\/a><\/p>\n
Figure 17 – Creating a new data set link<\/span><\/em><\/p>\n
From the Integration data set blade, select Azure Data Lake Storage Gen2<\/strong> and click on the Continue<\/strong> button at the bottom.<\/p>\n
\"New<\/a><\/p>\n
Figure 18 – New Azure Data Lake Storage Gen2 integration data set<\/span><\/p>\n
On the data format selection blade, select JSON<\/strong> and click on the Continue <\/strong>button at the bottom.<\/p>\n
\"Selecting<\/a><\/p>\n
Figure 19 – Selecting JSON format for our integration data set<\/em><\/span><\/p>\n
On the Set properties<\/strong> blade, select the linked service you\u2019ve previously created to your Azure Data Lake Storage location, specify the path to the folder where your messages data is being exported and click on the OK<\/strong> button at the bottom.<\/p>\n
\"Selecting<\/a><\/p>\n
Figure 20 – Selecting files location for our integration data set<\/em><\/span><\/p>\n
Repeat the same process, but this time for the Users data set. Make sure you set the path to point to the Users folder. At this stage, you should have a Data Flow that looks like the following:<\/p>\n
\"Data<\/a><\/p>\n
Figure 21 – Data flow with just two sources<\/span><\/em><\/p>\n
Next, we need to combine the two data sources. To do so, click on the +<\/strong> sign on the right of the Messages source in the workbench. From the list of actions, select Join<\/strong>.<\/p>\n
\"Adding<\/a><\/p>\n
Figure 22 – Adding a Join Action to a data flow<\/em><\/span><\/p>\n
In the editor panel at the bottom, select the Users source <\/strong>in the Right Stream<\/strong> drop down. For the Join condition<\/strong> make sure to select the pUser<\/strong> field for both sources.<\/p>\n
\"Configuring<\/a><\/p>\n
Figure 23 – Configuring a Join Action on the pUser Field<\/em><\/span><\/p>\n
Because both data sets will contain fields that are named the same, running the flow as-is would throw errors complaining about duplicate columns (e.g. pageRow). To remediate to the issue, we will need to add a select action after the Join action to skip duplicate columns since we do not need them both for our scenario. In the workbench, beside the Join action added above, click on the + <\/strong>sign and select the Select<\/strong> action from the list.<\/p>\n
\"Adding<\/a><\/p>\n
Figure 24 – Adding a Select Action to a data flow<\/em><\/span><\/p>\n
Simply leave the default values, which by default will skip the duplicate columns.<\/p>\n
\"Configuring<\/a><\/p>\n
Figure 25 – Configuring the Select Action<\/span><\/em><\/p>\n
The last step in our data flow is to determine where to store the combined data. In our case, we will be selecting a new workspace as the data sink. In the workbench, beside the newly added Select action, click on the +<\/strong> sign and select Sink <\/strong>from the list.<\/p>\n
\"Adding<\/a><\/p>\n
Figure 26 – Adding a Sink Action to a data flow<\/em><\/span><\/p>\n
In the editor panel at the bottom, select Workspace DB<\/strong> from the Sink type list, select the default<\/strong> database from the database list, and type in Results<\/strong> as the name of the table. Note that this will create a new table named Results in the default database and store the combined data sets in it.<\/p>\n
\"Configuring<\/a><\/p>\n
Figure 27 – Configuring the Sink Action for our data flow<\/em><\/span><\/p>\n
This completes the steps required to combine our data sets using a Data flow activity. You can now publish your pipeline and execute it. Once it successfully completes, you can follow the steps from the previous section (Option 1) to check the merged data using the SELECT TOP 100 ROWS<\/strong> query.<\/p>\n
Conclusion<\/h2>\n
In this article, we have shown you two ways of combining data sets extracted from Microsoft Graph Data Connect using Azure Synapse Analytics. While developers may tend to prefer the Spark Notebook approach (Option 1) over the low-code approach described in Option 2, note that both approaches will result in the same outcome. You should choose the approach you and your organization are more familiar with. The data we have exported and combined could then be connected to a Power BI dashboard using its Azure Synapse Analytics connector to generate valuable Insights and Analysis for your organization.<\/p>\n","protected":false},"excerpt":{"rendered":"
This article covers the process of combining two data sets extracted via an Azure Synapse pipeline using Microsoft Graph Data Connect (MGDC).<\/p>\n","protected":false},"author":72603,"featured_media":9451,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"footnotes":""},"categories":[3],"tags":[61,161,31],"class_list":["post-9391","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-microsoft-graph","tag-azure","tag-azure-synapse","tag-microsoft-graph-data-connect"],"acf":[],"blog_post_summary":"
This article covers the process of combining two data sets extracted via an Azure Synapse pipeline using Microsoft Graph Data Connect (MGDC).<\/p>\n","_links":{"self":[{"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/posts\/9391","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/users\/72603"}],"replies":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/comments?post=9391"}],"version-history":[{"count":0,"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/posts\/9391\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/media\/9451"}],"wp:attachment":[{"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/media?parent=9391"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/categories?post=9391"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/microsoft365dev\/wp-json\/wp\/v2\/tags?post=9391"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}