Late Binding vs. Early Binding in ADF Data Flows

When transforming cloud data at scale in ADF, Mapping Data Flows allows Data Engineers to graphically design an execution plan of data transformation routines that will execute against Spark. The inherent nature of complex cloud-based big data is that the data will be messy and will frequently change shape and form.

We have built-in features in Azure Data Factory to make it easy for Data Engineers to build robust ETL projects in ADF that can transform massive amounts in a flexible way. without writing code. To build your transformations in a way that can account for changes in data shapes, you will build your data flows in a way that exploits what we call “late binding”.

In many cases, you will build ETL projects for your business that require well-defined schemas with data sources that do not frequently change. In those cases, you can use traditional early-binding in your data flows. These are the characteristics of early binding:

1. A well-defined schema in your dataset
2. The data shape for your data sources does not frequently change
3. Your transformation expressions use explicit column names from your datasets
4. Heavy use of Select transformations to manipulate schemas manually
5. Heavy use of manual mapping in Sink and Select transformations

An example set of Aggregation expressions that utilize early-binding would be these expressions below that I use in my baseball data flow to generate offensive statistics based on columns that I expect to be present in my source data:

For column mapping, I know exactly which fields I wish to map in my destination and I want to control the logical names, so I’m using manual mapping:

As your data flows become more complex, include flexible schemas, and utilize late-binding, you’ll use either auto-mapping to generate output for all late-arriving columns (also known as “drifted”) or rule-based mapping.

To build data flows that are more resilient to change, you can choose to leverage features for late binding. It is important to keep in mind that when utilizing these late-binding techniques, your logical flow may look a little more complex and requires you to work with your data in a way that is patterns-based.

1. Use datasets that are schemaless. Your dataset is used in the Source transformations to define the shape of your data as the data is acquired into a Spark data frame. ADF datasets do not require schemas and you can take advantage of that to work with flexible data sources. Any column that ADF sees in your source that is not defined in your dataset is labeled as “drifted”.
2. Build transformations that use patterns instead of named columns. Writing expressions that require explicitly named columns from a dataset are using “early binding”. Instead, use patterns and rules that look for data by position, by naming patterns and data types.
3. Late binding utilizes ADF’s schema drift features, so you’ll want to make sure you’ve selected “schema drift” on your source and sinks. This way, ADF can pass through all newly-detected columns.

Here is an example of a Derived Column where I demonstrate pattern matching. From top-to-bottom:

1. I trim every String column, keeping the column name the same
2. I at 1.2 to the columns present at ordinal positions 2 and 3
3. I look for any column with the words “Customers” followed by a number using regular expressions. When that pattern is found, I generate a new column by appending “_customer” to the original column name. “$$” means that I am keeping the original value.

As your datasets become larger and wider, utilize rules in your Sink mapping like this example below. Here, I’m mapping any stat that contains “H” to the same target name in my destination table. I’m also including “SH” as a new column called “Sacrifices”.

These are examples of late binding that depend upon patterns and rules to define my transformation and mapping intent when I have data sources that are wide and change frequently.

For more details in these areas in ADF, read about:

1. Data Flow schema drift
2. Data Flow column patterns

Reduce Execution Time for Data Flow Activities in ADF Pipelines

In ADF Mapping Data Flows, there are 2 working modes: Debug mode and Pipeline mode.

Debug mode is active when you turn on the Data Flow debug switch and the light is green, showing debug as active. You will also see the Data Preview pane at the bottom of your transformation panel light-up with a green light. This will turn on the ability to interactively preview your data as you build your transformations in real-time in the data flow UI.

You can also execute pipelines with data flow activities against that same live Azure IR. To do this, you click “Debug” from the pipeline. Your pipeline will execute immediately and you can view your data flow activity in the bottom panel.

The other mode of working in Data Flows is from an operationalized pipeline. That is, a pipeline that has been scheduled from a trigger and runs against the live ADF service. Typically, you do this as your final testing step and then schedule your pipeline for normal operations. You can also get to this mode interactively from the pipeline screen (see above) by clicking Add Trigger > Trigger Now.

When you use Trigger Now, or a pipeline scheduled run, you no longer are using the debug Azure IR compute environment. Instead, you will use the Azure IR that is selected in each of your Data Flow activities.

Each of your data flow activities can execute from a different configuration of Azure IR. That means that you can apply more or fewer resources to each execution. You define the size of the compute environment from the Data Flow properties in the Azure IR:

If you set the TTL, ADF will spin-up and maintain a pool of resources that can be reused for that period of time. Every time that you request a data flow activity execution against that same Azure IR, ADF can load the cluster compute and job execution on a warm VM, reducing the overall execution time of your data flow. If you do not specify a TTL, then ADF will always spin-up new compute clusters on every execution.

This model ADF presents is very economical because you do not have to maintain an always-running Spark cluster to serve your ETL needs. With ADF, the compute is ephemeral and only present when it is needed.

The average start-up time for a just-in-time cluster is 5-7 minutes. If you have the need to lower that start-up time, then utilize the TTL setting. This will still require that initial load time to provision a pool of resources for that Azure IR. But each subsequent execution that occurs within that TTL window will have less acquisition time required.

In my example above, I executed my JSON Data Flow first against an Azure IR without TTL and it took 13 mins. Then, I executed it again, this time using a TTL of 10 mins. In this case, the data flow took almost half that time. Because I used a 10 min TTL, the cluster spun-up from warm VMs and I only had to pay for an extra 10 min TTL.

On the monitoring view, I am able to see the difference in cluster acquisition time from my Azure IR with no TTL vs. with TTL. Just keep in mind that your first activity will spin-up the resource pool which requires that same 5-7 minute build time.

ADF Data Flows: Distinct Rows

Below is a method to use in ADF’s Mapping Data Flows to reduce the data stream in your data flows to only include distinct rows. This sample is available as a pipeline template here.

1. Choose your source. This will work with any type of source.
2. Add an Aggregate transformation. Here I call it “DistinctRows”.
3. In the group by setting, you need to choose which or column or combination of columns will make up the key(s) for ADF to determine distinct rows. In this simple demo, I’ll just pick “movie” as my key column:
4. The inherent nature of the aggregate transformation is to block all metadata columns not used in the aggregate. But here, we are using the aggregate to filter out non-distinct rows, so we need every column from the original dataset.
5. To do this, go to the aggregate settings and choose Column Pattern.
6. Here, you will need to make a choice between including the first set of values from the duplicate rows, or the last. Essentially, choose which row you want to be the source of truth. 
7. That’s really it. That’s all you need to do to find distinct rows in your data.
8. The rest of this data flow sample is comparing the distinct row to the original set of data. You can keep the other transformation streams in this sample data flow so that you can compare the original data with the distinct row to make sure it is behaving the way you expect.
9. I created a copy of the original source data by using New Source and then renamed that stream as “OriginalData” by aliasing it with the Select transformation on the bottom row.
10. The row counts are just Aggregate transformations.
11. To create row counts go to Aggregate settings and leave the group by empty.
12. In the aggregate function, use the function count(1). This will create a running count of every row.

ETL with ADF: Convert Pig to Data Flows

Here’s a brief posting on taking an ETL script written in Pig. I took an ETL example using Pig from the Hortonworks tutorials site and migrating it to ADF using Mapping Data Flows. It took me approximately 10 minutes to design it.

What we’ll do in this demo is to create projections from two separate CSV files sitting in our data lake, aggregate the hours logged and miles logged by drivers, then join the datasets together for a complete report that is stored back in a lake folder. The full tutorial in Pig is here. You can download the source files to play along at home here.

I was able to take this 13-line Pig script and build it visually, without any coding, in ADF Data Flows in about 10 minutes. All it took was configuring 6 transformations on the design surface and I didn’t need to bother with Hadoop, MapReduce, Spark, Databricks or any other Big Data tech. Just define your data transformation intent in the ADF design surface and Data Flows handles the rest by spinning-up a Spark environment JIT, executing your transformations at scale, then tearing down the cluster for you.

drivers = LOAD 'drivers.csv' USING PigStorage(',');
raw_drivers = FILTER drivers BY $0>1;
drivers_details = FOREACH raw_drivers GENERATE $0 AS driverId, $1 AS name;
timesheet = LOAD 'timesheet.csv' USING PigStorage(',');
raw_timesheet = FILTER timesheet by $0>1;
timesheet_logged = FOREACH raw_timesheet GENERATE $0 AS driverId, $2 AS hours_logged, $3 AS miles_logged;
grp_logged = GROUP timesheet_logged by driverId;
sum_logged = FOREACH grp_logged GENERATE group as driverId,
SUM(timesheet_logged.hours_logged) as sum_hourslogged,
SUM(timesheet_logged.miles_logged) as sum_mileslogged;
join_sum_logged = JOIN sum_logged by driverId, drivers_details by driverId;
join_data = FOREACH join_sum_logged GENERATE $0 as driverId, $4 as name, $1 as hours_logged, $2 as miles_logged;
dump join_data;

Here are the steps to create this as an ADF Mapping Data Flow so that your end result will look something like this:

1. Create Delimited Text datasets for drivers.csv and timsesheet.csv. Switch on headers in the dataset configuration and import schema from source.
2. Create a new mapping data flow.
3. Switch on Data Flow Debug mode.
4. Add a Source transformation to the data flow canvas.
5. Select the DriversCSV dataset that you created in step 1.
6. Set the projection data types to match these formats:
7. Add a 2nd Source transformation and select the timesheetCSV dataset from step 1 and set the projection data types as such:
8. Validate that your data looks correct in the data preview tab on both sources. You should see something like this for each:
9. Make sure that your data types and data look correct. We’ll use the driverID as the PK/FK for the Join and the hours logged and miles logged for aggregations.
10. Let’s build the aggregations next. Add an aggregate transformation after your timesheet source transformation.
11. Name your aggregate transformation stream to be “SumHoursAndMiles”.
12. Configure the group-by field to be “driverId” because we’ll calculate the hours and miles per driver.
13. In the aggregates settings, create 2 new fields using the sum() function to look like this:
14. Use the data preview tab on the Aggregate transformation to validate that the data is summing properly:
15. You should see a sum total for your 2 new fields next to each driver ID.
16. Now we can join the 2 streams together so that we can see the full driver details along with the aggregated sums.
17. Add a Join transformation to your driversCSV source transformation.
18. Configure the Join for inner join on the “driverID” column:
19. There are a few duplicate columns on both streams, so let’s prune out the duplicate columns by using a Select transformation to curate the metadata.
20. In the Select transformation, all you have to do is click “Skip Duplicates” for both input and output:
21. Now you have a clean projection that you can sink to your data lake.
22. Add a Sink transformation that writes to a folder dataset in Blob Store or ADLS Gen 2.
23. Keep auto-map on the Sink and click data preview to make sure your data is looking correct. You should now see driver details along with their aggregate values:
24. That’s it for designing your data flow from the Pig example. If you want to execute this to produce output files into your lake folder sink, create a pipeline in ADF and add a Data Flow activity.
25. The best practice is to point the data flow activity to your new data flow and test it using the Debug button in the pipeline canvas before you set a triggered schedule for it. This way, you can verify that the data is being written to your lake folder properly and you won’t have to wait for the Databricks cluster spin-up time.

ADF Mapping Data Flows Parameters

Using Azure Data Factory Mapping Data Flows, you can make your data transformations flexible and general-purpose by using parameters. Use Data Flow parameters to create dynamic transformation expressions and dynamic contents inside of transformation settings. The online documentation for Data Flow parameters can be found here.

1. To get started, open a Data Flow and click on the Parameters tab
2. Here is where you can create and manage the Parameters that you will use inside of your data flows. You have the option of setting an optional default value here.
3. You can also create parameters from the Expression Builder.
4. You can use those parameters in your expressions to make dynamic transformations. Access the read-only parameter values inside your expressions.
5. Use parameters to make the settings in your source & sink transformations dynamic. Make your data flows dynamic based upon values sent in from the pipeline execution.
6. Click inside configurable settings fields and you’ll see the “Add dynamic content” link pop-up. Click to launch the expression builder and enter static values or dynamic expressions that utilize your parameters.
7. Now that you have your data flow designed with parameters, you can test it from a pipeline debug run. Click on the Execute Data Flow activity and you’ll see a Parameters tab on the bottom settings panel. Click inside the Value to set a static or dynamic value in either Data Flow Expression language or using the pipeline expressions (param must be type string).

 

 

Dynamic SQL Table Names with Azure Data Factory Data Flows

You can leverage ADF’s parameters feature with Mapping Data Flows to create pipelines that dynamically create new target tables. You can set those table names through Lookups or other activities. I’ve written a very simply post below on the tools you’ll need to do this:

1. Create a new ADF pipeline.
2. Switch on “Data Flow Debug”.
3. Create 2 new datasets. We’ll use Azure SQL DB for this demo, so create 2 Azure SQL DB datasets.
4. 
5. Go to the Parameter tab in the dataset and add a parameter for “tablename” as a sring.
6. In both of the datasets, do not select a table name. Instead, leave the table name blank and click in the table name field to display the “Add Dynamic Content” link.
7. 
8. Select the “tablename” parameter for the value in the Table field.
9. You’ll do this for both datasets. We’ll use 1 dataset for the source and the other for the sink destination.
10. Add an Execute Data Flow activity to your pipeline canvas and create a new data flow.
11. 
12. Inside the data flow canvas, select the dataset for the source table.
13. Add a Sink transformation directly after the source.
14. Choose the dataset for the destination table.
15. Back in the pipeline, click on the Execute Data Flow activity.
16. 
17. In the Settings tab, you’ll see a prompt for the values for the incoming and outgoing table names.
18. For this demo, I just typed in static text. I have an existing table in my SQL DB called “dbo.batting1”. I want to copy it as “dbo.batting2”.
19. This pipeline will copy it as a different name in the same database.
20. In a real-world scenario, you will set these dataset parameters via values from Lookup or other activities that change dynamically.
21. Click on “Debug” to test your pipeline.
22. After the debug run is executed, you should now see a new table in your Azure SQL DB with the name that you provided in the 2nd dataset parameter.

ADF Mapping Data Flows: Optimize for File Source and Sink

I’m going to use this blog post as a dynamic list of performance optimizations to consider when using Azure Data Factory’s Mapping Data Flow. I am going to focus this only to files. I will post subsequent articles that list ways to optimize other source, sinks, and data transformation types. As I receive more good practices, feedback, and other performance tunings, I will update this article accordingly.

Here is Azure SQL DB Optimizations for ADF Data Flows

Here is Azure SQL DW Optimizations for ADF Data Flows

Optimizations to consider when using ADF Mapping Data Flows with files

NOTE: When you are designing and testing Data Flows from the ADF UI, make sure to turn on the Debug switch so that you can execute your data flows in real-time without waiting for a cluster to warm up.

You can control how many partitions that ADF will use

1. On each Source & Sink transformation, as well as each individual transformation, you can set a partitioning scheme.
2. For smaller files, you may find selecting “Single Partition” can sometimes work better and faster than asking Spark to partition your small files.
3. If you do not have enough information about your source data, you can choose “Round Robin” partitioning and set the number of partitions.
4. If you explore your data and find that you have columns that can be good hash keys, use the Hash partitioning option.

File naming options

1. The default nature of writing transformed data in ADF Mapping Data Flows is to write to a dataset that has a Blob or ADLS Linked Service. You should set that dataset to point to a folder or container, not a named file.
2. Data Flows use Azure Databricks Spark for execution, which means that your output will be split over multiple files based on either default Spark partitioning or the partitioning scheme that you’ve explicitly chosen.
3. A very common operation in ADF Data Flows is to choose “Output to single file” so that all of your output PART files are merged together into a single output file.
4. However, this operation requires that the output reduces to a single partition on a single cluster node.
5. Keep this in mind when choosing this popular option. You can run out of cluster node resources if you are combining many large source files into a single output file partition.
6. To avoid this, you can keep the default or explicit partitioning scheme in ADF, which optimizes for performance, and then add a subsequent Copy Activity in the pipeline that merges all of the PART files from the output folder to a new single file. Essentially, this technique separates the action of transformation from file merging and achieves the same result as setting “output to single file”.

Increase size of your compute engine in Azure Integration Runtime

1. Increase the number of cores, which will increase the number of nodes, and provide you with more processing power to process your file operations.
2. Try “Compute Optimized” and “Memory Optimized” options.

ADF Mapping Data Flows: Optimize for Azure SQL Data Warehouse

I’m going to use this blog post as a dynamic list of performance optimizations to consider when using Azure Data Factory’s Mapping Data Flow. I am going to focus this only to Azure SQL DW. I will post subsequent articles that list ways to optimize other source, sinks, and data transformation types. As I receive more good practices, feedback, and other performance tunings, I will update this article accordingly.

Here is Azure SQL DB Optimizations for ADF Data Flows.

Optimizations to consider when using ADF Mapping Data Flows with Azure SQL DW

NOTE: When you are designing and testing Data Flows from the ADF UI, make sure to turn on the Debug switch so that you can execute your data flows in real-time without waiting for a cluster to warm up.

You can match Spark data partitioning to your source database partitioning based on a database table column key in the Source transformation

1. Go to “Optimize” and select “Source”. Set either a specific table column or a type in a query.
2. If you chose “column”, then pick the partition column.
3. Also, set the maximum number of connections to your Azure SQL DW. You can try a higher setting to gain parallel connections to your database. However, some cases may result in faster performance with a limited number of connections.

Set Batch Size and Query on Source

1. Setting batch size will instruct ADF to store data in sets in memory instead of row-by-row. It is an optional setting and you may run out of resources on the compute nodes if they are not sized properly.
2. Setting a query can allow you to filter rows right at the source before they even arrive for Data Flow for processing, which can make the initial data acquisition faster.

Use staging to load data in bulk via Polybase

1.  In order to avoid row-by-row processing of your data floes, set the “Staging” option in the Sink settings so that ADF can leverage Polybase to avoid row-by-row inserts into DW. This will instruct ADF to use Polybase so that data can be loaded in bulk.
2. When you execute your data flow activity from a pipeline, with Staging turned on, you will need to select the Blob store location of your staging data for bulk loading.

Set Partitioning Options on your Sink

1. Even if you don’t have your data partitioned in your destination Azure SQL DW tables, go to the Optimize tab and set partitioning.
2. Very often, simply telling ADF to use Round Robin partitioning on the Spark execution clusters results in much faster data loading instead of forcing all connections from a single node/partition.

Set isolation level on Source transformation settings for SQL datasets

1. Read uncommitted will provide faster query results on Source transformation

Increase size of your compute engine in Azure Integration Runtime

1. Increase the number of cores, which will increase the number of nodes, and provide you with more processing power to query and write to your Azure SQL DW.
2. Try “Compute Optimized” and “Memory Optimized” options

Disable indexes on write

1. Use an ADF pipeline stored procedure activity prior to your Data Flow activity that disables indexes on your target tables that are being written to from your Sink.
2. After your Data Flow activity, add another stored proc activity that enabled those indexes.

Increase the size of your Azure SQL DW

1. Schedule a resizing of your source and sink Azure SQL DW before you run your pipeline to increase the throughput and minimize Azure throttling once you reach DWU limits.
2. After your pipeline execution is complete, you can resize your databases back to their normal run rate.

ADF Mapping Data Flows: Optimize for Azure SQL Database

I’m going to use this blog post as a dynamic list of performance optimizations to consider when using Azure Data Factory’s Mapping Data Flow. I am going to focus this only to Azure SQL DB. I will post subsequent articles that list ways to optimize other source, sinks, and data transformation types. As I receive more good practices, feedback, and other performance tunings, I will update this article accordingly.

Here is Azure SQL DW Optimizations for ADF Data Flows.

Optimizations to consider when using ADF Mapping Data Flows with Azure SQL DB

NOTE: When you are designing and testing Data Flows from the ADF UI, make sure to turn on the Debug switch so that you can execute your data flows in real-time without waiting for a cluster to warm up.

Partition your source query

1. Go to “Optimize” and select “Source”. Set either a specific table column or a type in a query.
2. If you chose “column”, then pick the partition column.
3. Also, set the maximum number of connections to your Azure SQL DB. You can try a higher setting to gain parallel connections to your database. However, some cases may result in faster performance with a limited number of connections.
4. Your source database tables do not need to be partitioned.
5. Setting a query in your Source transformation that matches the partitioning scheme of your database table will allow the source database engine to leverage partition elimination.
6. If your source is not already partitioned, ADF will still use data partitioning in the Spark transformation environment based on the key that you select in the Source transformation.

Set Batch Size and Query on Source

1. Setting batch size will instruct ADF to store data in sets in memory instead of row-by-row. It is an optional setting and you may run out of resources on the compute nodes if they are not sized properly.
2. Setting a query can allow you to filter rows right at the source before they even arrive for Data Flow for processing, which can make the initial data acquisition faster.
3. If you use a query, you can add optional query hints for your Azure SQL DB, i.e. READ UNCOMMITTED

Set isolation level on Source transformation settings for SQL datasets

1. Read uncommitted will provide faster query results on Source transformation

Set Sink Batch Size

1.  In order to avoid row-by-row processing of your data floes, set the “Batch size” in the sink settings for Azure SQL DB. This will tell ADF to process database writes in batches based on the size provided.

Set Partitioning Options on your Sink

1. Even if you don’t have your data partitioned in your destination Azure SQL DB tables, go to the Optimize tab and set partitioning.
2. Very often, simply telling ADF to use Round Robin partitioning on the Spark execution clusters results in much faster data loading instead of forcing all connections from a single node/partition.

Increase size of your compute engine in Azure Integration Runtime

1. Increase the number of cores, which will increase the number of nodes, and provide you with more processing power to query and write to your Azure SQL DB.
2. Try “Compute Optimized” and “Memory Optimized” options

Disable indexes on write

1. Use an ADF pipeline stored procedure activity prior to your Data Flow activity that disables indexes on your target tables that are being written to from your Sink.
2. After your Data Flow activity, add another stored proc activity that enabled those indexes.

Increase the size of your Azure SQL DB

1. Schedule a resizing of your source and sink Azure SQL DB before your run your pipeline to increase the throughput and minimize Azure throttling once you reach DTU limits.
2. After your pipeline execution is complete, you can resize your databases back to their normal run rate.