Microsoft Fabric – NHS data processing… part 3, landing the data in a lakehouse

Why?

Some people don’t use a landing zone and load (or merge) directly into tables, but for me I’d always recommend you have a landing area (or whatever you want to call it). Somewhere in your architecture that you can write files containing your data extracts. Often in the cloud this will and should be a data lake due to the speed and flexibility. In Fabric that means we need to create a lakehouse, as this is the only place you can write files to within OneLake.

In our example we’re pulling data files from a website, but we’d want to do this even if the data is extracted from a relational database source such as SQL server or PostgreSQL. Why? Well, this simplifies the import process, but also gives you a history of data files which can then be used (and re-used) for any reloads, refreshes, testing or bug checking later on if anything crops up… all without needing to go back to the source system again. Obviously going back and querying the source each time to check stuff could be problematic, slow, and/or affect performance for system users, especially if you need to check a lot of data.

We’ll go one step further though and in addition to landing the file in the lake in a nice ordered structure, we’ll also load each incoming file into a table, overwriting the schema and contents as we do. This means we can do some quick queries and checks on the data if we need to, it also means later on if we want or need to query landing data from a warehouse. You could take the files and merge them into straight into the staging lakehouse ‘history’ tables without creating a table in the landing zone though, but I’d suggest it can be useful for little to no performance cost.

Here’s a little screenshot of the various lake houses and warehouses we’ve got. We’re going to look at the ‘LandingLakehouse’ shown below. Note that as mentioned in the previous post I’ve grouped these all together in a folder in the workspace called ‘Databases’…

Opening the ‘LandingLakehouse’ we’ve got the following structure. Various tables and the files which the process has downloaded to the lake. As you can see the process has created sub folders to store each of the different datasets files to clean things nice, clean and ordered. Also note that the dataset folders each have 2 sub folders, one for zipped files and another for unzipped files. We’ll download our zipped files into the relevant folder then unzip them into the other folder 🫡

One other thing to note is that I’ve created a shortcut to the ‘ControlWarehouse’ ‘DataSources’ table so I can query that table from the lakehouse. This will come in useful later… on a side note, there are no shortcuts to lake houses in any of the warehouses as shortcuts can only be created in a lakehouse. In a warehouse though, you can just query the lakehouse SQL endpoints so there’s no need for shortcuts in warehouses…

As I mentioned earlier, we’re using some pyspark to download the file from the website instead of using data factory due to the connection parameterisation limitation. Ideally I would just use the COPY activity in data factory and to be honest, just take it as a given that when that feature is released this bit of code will be redundant. Obviously if there’s some other reason to use pyspark for landing then of course use it, but normally for simple copying always take the path of least resistance – in this case it’s likely cheaper too, but for now, pyspark it is…

Show me the code…

In our ‘download to landing’ notebook I’ve written this small function which we can use to download a file and write it to the lake where we want to:-

# Downloads a file from the specified URL to the location specified in the lake
#
# Test example:
#
# url = "https://www.england.nhs.uk/statistics/wp-content/uploads/sites/2/2024/02/Full-CSV-data-file-Dec23-ZIP-3569K-58522.zip"
# location_to_write_file_to_in_lake = "/lakehouse/default/Files/referral_to_treatment/zipped_files/Full-CSV-data-file-Dec23-ZIP-3569K-58522.zip"
# download_file(url, location_to_write_file_to_in_lake)

def download_file(url, location_to_write_file_to_in_lake):
    import requests
    import pathlib 
    import io

    # Create the web request and try and get a response
    http_response = requests.get(url)

    # For debugging, print the location the file gets written to
    print(f"Starting download from '{url}'")
    print(f"Intended destination: '{location_to_write_file_to_in_lake}'")

    # Write a copy of the file to the lake    
    pathlib.Path(location_to_write_file_to_in_lake) \
        .write_bytes(io.BytesIO(http_response.content) \
            .getbuffer() \
            .tobytes())

    # For debugging, print the location the file gets written to
    print(f"Completed downloaded from '{url}'")
    print(f"Downloaded to: '{location_to_write_file_to_in_lake}'")

Also, to unzip the files I’m using another little function. I could have directly used the ‘ZipFile’ functions on their own as this is quite short, but I like to wrap code like this into small ‘single block’ functions even if its maybe 2 or 3 lines of code in the ‘flesh’. That’s just me, feel free to do whatever you prefer 🫡

# Small function to help unzip files

def unzip_file(zipped_file_path, path_to_unzip_files_to):    
    import zipfile

    # Get the zipped file
    zipped_file = zipfile.ZipFile(zipped_file_path)
    
    # Example zipped_file.extractall(f"/lakehouse/default/{path_to_unzip_files_to}/{file_name}")
    zipped_file.extractall(path_to_unzip_files_to)

Lastly… whilst this is not part of the landing notebook itself, I used this little function to scrape the web page to get all the links to the files I wanted to download. This could of course be expanded upon I guess to automatically scrape several pages and follow links to other downloads, maybe on a schedule. But for now, this is just something I manually ran once to get the list of file links and then I just loaded those into the DataSources table in the ‘ControlWarehouse’ for processing 👌

# Scrapes the specified url for file download links

def scape_url_for_file_links(url_to_scrape, file_extension_filter = (".csv", ".zip", ".pdf", ".txt")):    
    # Use the requests library to get the HTML content of the webpage
    import requests
    response = requests.get(url)

    # Parse the HTML content using BeautifulSoup
    from bs4 import BeautifulSoup
    soup = BeautifulSoup(response.text, "html.parser")

    # Find all links on the page that point to files
    file_links = []

    for link in soup.find_all("a"):
        # Find the href attribute in the link
        href = link.get("href")

        # Filter for only the extensions we want
        if href and href.endswith(file_extension_filter):        
            obj = dict(link_url = href, link_text = link.get_text())            
            file_links.append(obj)

    return file_links

Not to forget, once we’ve downloaded the files we can use these functions to load them into a table. The first function just ‘cleans’ the column names as a lot of NHS CSV files have column names with things like spaces, dashes and other characters that don’t play well with lakehouse table column naming rules. The next function actually loads the file into a table…

# Removes or replaces 'invalid' characters in column names in a data 
# frame ready for insert into a delta table

def clean_column_names(data_frame, replace_characters = True):
    from pyspark.sql.functions import col

    # Are we replacing characters with alternatives?    
    if replace_characters:
        # Replace invalid characters with valid ones
        df = (
            data_frame.select([col(column_name)
                .alias(column_name
                    .replace(" ", "_")            
                    .replace("-", "_")
                    .replace("&", "and")
                ) for column_name in data_frame.columns]
            )
        )

    # No, we're just removing the characters only!
    else:
        # Only remove the characters, don't replace them with anything
        df = (
            data_frame.select([col(column_name)
                .alias(column_name
                    .replace(" ", "")            
                    .replace("-", "")
                    .replace("&", "")
                ) for column_name in data_frame.columns]
            )
        )

    return df

# Loads the CSV file into a data frame

def load_data_file_into_table(file_format, path_to_data_file, table_name, header):
    # Read the data into a data frame
    df = spark.read.format("csv").option("header", header).load(path_to_data_file)

    # Since the tables only allow alphanumeric characters and underscores for column names, we might
    # potentially need to rename some columns. So here we look for some common characters and just
    # replace. Note that you might need to add more characters to this code if your CSV headers
    # contain other 'invalid' or unsupported characters
    final_df = clean_column_names(df, False)
    
    # Lastly create the table from the data frame - note that we want to overwrite any existing table!
    final_df.write.mode("overwrite").option("overwriteSchema", "true").format("delta").saveAsTable(table_name)

So, all this wrapped up in a notebook (with parameters for the URL etc…) gets called by the data factory pipeline in the ‘landing’ step and hey presto we’ve got a file downloaded and a table created in the landing lakehouse with the contents of the latest file we downloaded. Note the ‘overwrite’ as we only want the latest file data in the table in landing, we’re only building a history table of all the data in the ‘staging’ lakehouse!

So that is pretty much the landing zone, next time we’ll look at the staging area where we’ll merge this incoming data in landing into ‘history’ tables…