![\"Microsoft](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/microsoft-fabric-logo.png\"){kind=logo}
<\/p>\n<\/p>\n
The loading of data from a source system to target system has been well documented<\/a> over the years. My first introduction to an Extract, Transform and Load program was DTS<\/a> for SQL Server 7.0 in 1998.<\/p>\n In a data lake, we have a bronze quality zone that supposed to represent the raw data in a delta file format. This might include versions of the files for auditing. In the silver quality zone, we have a single version of truth. The data is de-duplicated and cleaned up. How can we achieve these goals using the Apache Spark engine in Microsoft Fabric? <\/p>\n Our manager has given us weather data to load into Microsoft Fabric. In the last article, we re-organized the full<\/b> load sub-directory to have two complete data files (high temperatures and low temperatures) for each day of the sample week. As a result, we have 14 files. The incremental<\/b> load sub-directory has two files per each day for 1369 days. The high<\/b> and low<\/b> temperature files for a given day each have a single reading. There is a total of 2,738 files. How can we create a notebook to erase existing tables if needed, rebuild the full load tables, and rebuild the incremental tables?<\/p>\n This use case allows data engineers to learn how to transform data using both Spark SQL and Spark DataFrames. The following topics will be explored in this article (thread).<\/p>\n I borrowed the image from Emily Riederers<\/a> blog since I really liked it. What is the difference between a full and incremental load?<\/font> A full load<\/b> means that all data is copied over every scheduled execution of the notebook. The main question to ask the business is “do you need auditing of the data”? If so, we need to keep the full files for each processing day. One great use case for this pattern is critical business process data. Let’s make believe we are a chocolate maker and we have a set of recipes for creating our chocolate items. We might want to know when the recipe was changed for the almond clusters?<\/p>\n A full load is great for small to medium datasets that do not often change. For extremely large datasets, start with a multi file full load for historical data given the retention period you are looking for. Then going forward, use a incremental files to keep the dataset up to date. In our proof of concept today, we never go back and update history. We will talk about merging datasets with incremental loading in a future thread will allow for this use case.<\/p>\n Fabric supplies the data engineer with a parameter cell<\/a>. We can see that the process<\/font><\/b> variable is currently set to full<\/b>. However, the Python code in the notebook was built to support erase<\/b>, incr<\/b> and full<\/b> operations (values). Additionally, we might want to debug<\/font><\/b> variable to be used to optionally display sample data and row counts. This variable supports both true<\/b> and false<\/b> values.<\/p>\n The above image shows the parameter cell we are using in this notebook. The image below shows the end state of the hive catalog. It will have four published tables.<\/p>\n The show tables<\/font><\/b> Spark SQL command lists both permanent and temporary tables.<\/p>\n After executing the above Spark SQL statement we see there are 3 tables in the spark session for each load type: bronze table, silver table and temporary table.<\/p>\n We are going to use the if statement to optionally execute code. If our process variable is set to erase<\/b>, the following code will execute.<\/p>\n Here is a quick overview of the above code. First, we grab a DataFrame containing all the lakehouse tables by calling the spark.sql<\/a> method with the correct Spark SQL statement. The collect<\/a> method of the DataFrame returns a list. Then for each table in the list, we execute a drop table<\/a> command to remove the tables from the hive catalog. The spark<\/b> variable is just a point to the Spark Session. We will be using this method many times in the future.<\/p>\n The above image shows all the tables have been removed from the hive catalog.<\/p>\n In this section, we are going over code that will be used to create bronze and silver tables using a full load pattern. If we try to create a table when it already exists, the program will error out. The first step<\/b> in our full load process is to remove the existing bronze and silver tables.<\/p>\n The second step<\/b> in our full load process is to read in the data files as a Spark DataFrame. To fix the data type issue we had in the prior article, we are going to pass a schema definition to the read<\/a> method. The createOrReplaceTempView<\/a> method is very powerful. It registers the Spark DataFrame as a temporary view.<\/p>\n The third step<\/b> is to use Spark SQL to transform the data. The create table as<\/a> command allows us to create a hive table with the resulting data from the select<\/a> statement. Another useful Spark SQL function is input_file_name<\/a>. The image below shows the full path of the processed file with some additional URL info. The split_part<\/a> function is very useful when parsing a delimited string and selecting a particular item in an array.<\/p>\n Other than adding additional fields such as folder name, file name, and insert date, the data in the bronze delta table reflects the raw data stored as files and folders. That is about to change when we create the silver delta table.<\/p>\n We are going to use a common table expression to create two derived tables: cte_high_temps<\/b> and cte_low_temps<\/b>. I purposely created the folders using a year, month, and day format. This is a sortable directory. To find the most recent data, we just need to select the data that has the maximum folder value. <\/p>\n Additionally, we want to combine the high temperate and low temperature reading for a given day. I am keeping the source folder name, the high temp file name and the low temp file name as lineage.<\/p>\n The image below shows the new tables in the hive catalog when executing the Spark Notebook with process equal to full<\/b>.<\/p>\n I am going to skip over the code snippets since the complete notebook is enclosed at the end of the article. The first image shows that for a given day, we have 7 entries for high temperatures and 7 entries for low temperatures in the bronze table.<\/p>\n The second image shows the record count from the bronze table. We have a-lot of duplicate data.<\/p>\n The third image that the silver table has one row per day. That row contains both high and low temperature readings.<\/p>\n The fourth image shows the silver table has the same number of rows equal to the number of distinct days.<\/p>\n In summary, the bronze table for the full load pattern can be quite large if we decide to keep a-lot of historical files.<\/p>\n In this section, we are going over code that will be used to create bronze and silver tables using a incremental load pattern. There is not historical component since we have data for each day. If we try to create a table when it already exists, the program will error out. The first step<\/b> in our incremental load process is to remove the existing bronze and silver tables.<\/p>\n The second step<\/b> in our incremental load process is to read in the data files as a Spark DataFrame. A schema is used to correctly type the data as well as a temporary view. This view allows us to use Spark SQL in steps three and four to transform and create the medallion tables.<\/p>\n The third step<\/b> creates the bronze table. Since no filtering and\/or transformations are being performed, the code almost matches the full load bronze table code.<\/p>\n The fourth step<\/b> creates our silver table. Again, we are using to common table expressions to separate and combine the high and low temperature data.<\/p>\n The image below shows two new tables for the incremental data.<\/p>\n I am going to skip over the code snippets since the complete notebook is enclosed at the end of the article. The first image shows that for a given day, we have a sub-folder and two files. One file is for a high temperature reading and one file is for a low temperature reading.<\/p>\n The second image shows that the number of files equals the number of days times two.<\/p>\n The third image shows that the two temperatures have been paired together as one data point.<\/p>\n The fourth image shows that the number of rows is equal to the number of days or 1369.<\/p>\n If we preview the data in the silver incremental table, we can grid looks like the power query interface. The correct data types are assigned to each column (field).<\/p>\n As a consultant, I find a-lot of companies skip steps in regards to system design, commenting code, peer review of code and code testing. Skipping these steps might result in a poor end product.<\/p>\n Going back to the medallion architecture<\/a>, we can modify the diagram to leave just the files as the data source. This is the only step I did not do in this article. In the next article, we are going to work on scheduling the Spark Notebook with a data pipeline. The SQL analytics endpoint will allow us to expose the existing bronze and silver tables as well as create gold views. This same end point will allow other tools like SSMS to connect to the one lake and use the data.<\/p>\n Fabric supplies the data engineer with access to the Apache Spark<\/a> engine to read, transform and write files. Once a file(s) is read into a DataFrame, we can publish a temporary view and perform all the transformation logic using Spark SQL. Yes, you do need to know a language like Python or Scala but remember to test your SQL statements with the %%sql magic command.<\/p>\n Is this code ready for production?<\/font><\/b> The answer is it all depends<\/b><\/font>. If you are looking for a full load and\/or a incremental append pattern, you are all set. This code will work with small<\/b> to medium<\/b> datasets. We are rebuilding the delta tables each time from scratch. The amount of processing time will increase linearly with the number of files. There is always the consideration of optimization of large datasets via partitioning<\/a> or v-order<\/a>. <\/p>\n In a future article, I will talk about how to handle large<\/b> datasets with partition and v-order. Additionally, for large datasets we do not want to rebuild the delta table each time. The MERGE<\/a> command can be used to update the data on a given key.<\/p>\n Let’s think about the sample dataset we were given. It splits data into both high and low temperatures which is crazy. But in real life, a file with longitude, latitude, date, time, and temperature is more appropriate. The sample rate does not matter as long as it is consistent between files. With this new file layout, we can collect data from a multiple set of GPS locations. A simple min and max query by date gives us our final data. <\/p>\nBusiness Problem<\/h4>\n
Technical Solution<\/h4>\n
\n
Architectural Overview<\/h4>\n
\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/incr-full-good.png\")
\nWe can see that a incremental load<\/b> may start with a full load.This load brings over any historical data and is depicted by the blue box. However, going forward we just bring over a new incremental file. This is depicted by the green box. Sometimes you might be collecting data from scratch. Thus, there is no history. This is what we are doing with the sample incremental weather data.<\/p>\nErase Tables<\/h4>\n
![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-01.png\")
<\/p>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-05.png\")
<\/p>\n\n%%sql\n--\n-- E2 - show hive + temp tables\n--\n\nshow tables;\n\n<\/pre>\n
![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-10.png\")
<\/p>\n\n#\n# E1 - remove all hive tables\n#\n\nif process == 'erase':\n tables = spark.sql('show tables;').collect()\n for table in tables:\n stmt = \"drop table \" + table.tableName\n ret = spark.sql(stmt)\n<\/pre>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-15.png\")
<\/p>\nFull Load<\/h4>\n
\n#\n# F1 - remove existing tables\n#\n\nif process == 'full':\n ret = spark.sql('drop table if exists bronze_weather_full;')\n ret = spark.sql('drop table if exists silver_weather_full;')\n<\/pre>\n\n#\n# F2 - load data frame\n#\n\n# load sub folders into df\nif process == 'full':\n \n # root path\n path = \"Files\/raw\/weather\/full\/\"\n\n # file schema\n layout = \"date timestamp, temp float\"\n\n # load all files\n df = spark.read.format(\"csv\") \\\n .schema(layout) \\\n .option(\"header\",\"true\") \\\n .option(\"recursiveFileLookup\", \"true\") \\\n .load(path)\n\n # convert to view\n df.createOrReplaceTempView('tmp_weather_full')\n<\/pre>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-70.png\")
<\/p>\n\n#\n# F3 - create bronze table (all files)\n#\n \n# spark sql\nstmt = \"\"\"\n create table bronze_weather_full as\n select \n date as _reading_date,\n temp as _reading_temp,\n current_timestamp() as _load_date,\n split_part(input_file_name(), '\/', 9) as _folder_name,\n split_part(split_part(input_file_name(), '\/', 10), '?', 1) as _file_name\n from \n tmp_weather_full \n\"\"\"\n\n# create table?\nif process == 'full':\n ret = spark.sql(stmt)\n<\/pre>\n
\n#\n# F4 - create silver table (lastest files)\n#\n \n# spark sql\nstmt = \"\"\"\n create table silver_weather_full as\n with cte_high_temps as\n (\n select * \n from bronze_weather_full as h \n where h._file_name = 'high_temps.csv' and\n h._folder_name = (select max(_folder_name) from bronze_weather_full)\n )\n ,\n cte_low_temps as\n (\n select * \n from bronze_weather_full as l\n where l._file_name = 'low_temps.csv' and\n l._folder_name = (select max(_folder_name) from bronze_weather_full)\n )\n select \n h._reading_date,\n h._reading_temp as _high_temp,\n l._reading_temp as _low_temp,\n h._load_date,\n h._folder_name,\n h._file_name as _high_file,\n l._file_name as _low_file\n from \n cte_high_temps as h \n join \n cte_low_temps as l\n on \n h._reading_date = l._reading_date\n order by\n h._reading_date desc\n\"\"\"\n\n# create table?\nif process == 'full':\n ret = spark.sql(stmt)\n<\/pre>\n
![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-20.png\")
<\/p>\nFull Table Testing<\/h4>\n
![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-25.png\")
<\/p>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-30.png\")
<\/p>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-35.png\")
<\/p>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-40.png\")
<\/p>\nIncremental Load<\/h4>\n
\n#\n# I1 - remove existing tables\n#\n\nif process == 'incr':\n ret = spark.sql('drop table if exists bronze_weather_incr;')\n ret = spark.sql('drop table if exists silver_weather_incr;')\n<\/pre>\n\n#\n# I2 - load data frame\n#\n\n# load sub folders into df\nif process == 'incr':\n \n # root path\n path = \"Files\/raw\/weather\/incremental\/\"\n\n # file schema\n layout = \"date timestamp, temp float\"\n\n # load all files\n df = spark.read.format(\"csv\") \\\n .schema(layout) \\\n .option(\"header\",\"true\") \\\n .option(\"recursiveFileLookup\", \"true\") \\\n .load(path)\n\n # convert to view\n df.createOrReplaceTempView('tmp_weather_incr')\n<\/pre>\n\n#\n# I3 - create bronze table (all files)\n#\n \n# spark sql\nstmt = \"\"\"\n create table bronze_weather_incr as\n select \n date as _reading_date,\n temp as _reading_temp,\n current_timestamp() as _load_date,\n split_part(input_file_name(), '\/', 9) as _folder_name,\n split_part(split_part(input_file_name(), '\/', 10), '?', 1) as _file_name\n from \n tmp_weather_incr\n\"\"\"\n\n# create table?\nif process == 'incr':\n ret = spark.sql(stmt)\n<\/pre>\n
\n#\n# F4 - create silver table (all files)\n#\n \n# spark sql\nstmt = \"\"\"\n create table silver_weather_incr as\n with cte_high_temps as\n (\n select * \n from bronze_weather_incr as h \n where h._file_name = 'high_temps.csv'\n )\n ,\n cte_low_temps as\n (\n select * \n from bronze_weather_incr as l\n where l._file_name = 'low_temps.csv'\n )\n select \n h._reading_date,\n h._reading_temp as _high_temp,\n l._reading_temp as _low_temp,\n h._load_date,\n h._folder_name,\n h._file_name as _high_file,\n l._file_name as _low_file\n from \n cte_high_temps as h \n join \n cte_low_temps as l\n on \n h._reading_date = l._reading_date\n order by\n h._reading_date desc\n\"\"\"\n\n# create table?\nif process == 'incr':\n ret = spark.sql(stmt)\n<\/pre>\n
![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-75.png\")
<\/p>\nIncremental Table Testing<\/h4>\n
![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-45.png\")
<\/p>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-50.png\")
<\/p>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-55.png\")
<\/p>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-60.png\")
<\/p>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/thread-003-65.png\")
<\/p>\n![](\"https:\/\/craftydba.com\/wp-content\/uploads\/2024\/02\/onelake-medallion-lakehouse-architecture-example.png\")
<\/p>\nSummary<\/h4>\n