{"id":9447,"date":"2024-02-23T18:47:36","date_gmt":"2024-02-23T18:47:36","guid":{"rendered":"https:\/\/craftydba.com\/?p=9447"},"modified":"2024-12-26T21:02:55","modified_gmt":"2024-12-26T21:02:55","slug":"thread-04-data-engineering-with-fabric","status":"publish","type":"post","link":"https:\/\/craftydba.com\/?p=9447","title":{"rendered":"Thread 04 \u2013 Data Engineering with Fabric"},"content":{"rendered":"

\"Microsoft<\/p>\n

Metadata Driven Pipelines<\/h2>\n

What is a metadata driven pipeline?<\/font> Wikipedia defines metadata<\/a> as “data that provides information about other data”. As a developer, we can create a non parameterized pipeline and\/or notebook to solve a business problem. However, if we have to solve the same problem a hundred times, the amount of code can get unwieldly. A better way to solve this problem is to store metadata in the delta lake. This data will drive how the Azure Data Factory and Spark Notebooks execute.<\/p>\n

Business Problem<\/h4>\n

Our manager has asked us to create a metadata driven solution to ingest CSV files from external storage into a Microsoft Fabric one lake. How can we complete this task?<\/p>\n

Technical Solution<\/h4>\n

We are going to go back to the drawing board for this solution. First, we have determined that the CSV files stored in the ADLS container are small to medium in size. Second, we are going to keep historical files around for auditing in the bronze layer. Third, the silver layer will present to the end users the most recent version of the truth (data). Since the datasets are tiny in nature, we are going to use a full load process.<\/p>\n

The following topics will be covered in this article (thread).<\/p>\n

    \n
  1. create metadata table<\/li>\n
  2. populate metadata table<\/li>\n
  3. create data pipeline<\/li>\n
  4. create credentials for storage<\/li>\n
  5. create parameters and variables<\/li>\n
  6. code for copy activity<\/li>\n
  7. code for notebook activity<\/li>\n
  8. test for 3 use cases<\/li>\n
  9. any future enhancements<\/li>\n<\/ol>\n

    Architectural Overview<\/h4>\n

    The architectural diagram for this metadata driven design is seen below.<\/p>\n


    \n
    There are three major components: Azure Data Lake Storage which is external to the Fabric service; Data Factory Pipeline Activities that read\/write files; and the Lake House storage with its medallion layers. Please note that medallion layers are either folders and\/or naming conventions to organize the data in logical buckets.
    \n
    \nA popular use case of a data lake is to collect data from various disconnected systems (producers<\/b>) and place the data into single location (hub<\/b>). However, there are always users (consumers<\/b>) that want to access the data using their favorite tools. That is why a data lake is considered a spoke<\/b> and hub<\/b> design.<\/p>\n

    <\/p>\n

    As a Microsoft Fabric designer we can use either data pipelines<\/b> or data flows<\/b> to move the data into the lake. I choose pipelines since they can be easily parameterized. At the transformation layer, we must choose between Spark notebooks<\/b> or data flows<\/b>. Again, I decided to pick notebooks since the code is portable to other platforms. Since the SQL endpoint does not support Spark views, we are going to have to add this gold layer using data warehouse views<\/b>.<\/p>\n

    Meta Data Tables<\/h4>\n

    A meta data driven design allows the developer to automate the copying of data into the lake (raw<\/b> files) as well as building the bronze<\/b> and silver<\/b> tables. The image below shows the meta data table viewed from the SQL endpoint.<\/p>\n

    <\/b>This is a spoiler alert.<\/b> Since the gold<\/b> tables are usually custom for each lake, we will be talking about Synapse Data Engineering and Data Warehouse design in our next article.<\/p>\n

    <\/p>\n

    The above image shows the end state of our article. The netflix, high temperature and low temperature raw files have been converted to both bronze and silver tables.<\/p>\n

    Lets focus on the meta data table right now with the following fields.<\/p>\n

      \n
    1. pipeline id – just an identifier<\/li>\n
    2. container name – same path in source and destination storage<\/li>\n
    3. file name – same file in source and destination storage<\/li>\n
    4. header flag – does the data file have a header line<\/li>\n
    5. delimiter string – how is the tuple divided into columns<\/li>\n
    6. table name – the root name of the table<\/li>\n
    7. schema string – infer schema or use this definition<\/li>\n<\/ol>\n

      The purpose of the this section is to show the developer how to drop table, create table, and insert data. These actions are required to create our meta data table and populate it.<\/p>\n

      \n%%sql\n--\n--  remove table\n--\n\ndrop table if exists meta_data;\n<\/pre>\n
      The above Spark SQL drops the meta data table and the below SQL code creates the table.<\/p>\n
      \n%%sql\n--\n--  create table\n--\n\ncreate table meta_data \n(\n    pipeline_id int,\n    container_nm string,\n    file_nm string,\n    header_flg boolean,\n    delimiter_str string,\n    table_nm string,\n    schema_str string\n);\n<\/pre>\n
      Insert meta data for the netflix dataset.<\/p>\n
      \n%%sql\n--\n--  netflix dataset\n--\n\ninsert into meta_data values (1, 'raw\/netflix', 'netflix_titles.csv', true, ',', 'netflix', '');\n<\/pre>\n
      Insert meta data for the high temperature dataset.<\/p>\n
      \n%%sql\n--\n--  weather - high temp dataset\n--\ninsert into meta_data values (2, 'raw\/weather', 'high_temps.csv', true, ',', 'high_temps', 'date timestamp, temp float');\n<\/pre>\n
      Insert meta data for the low temperature dataset.<\/p>\n
      \n%%sql\n--\n--  weather - low temp dataset\n--\n\ninsert into meta_data values (3, 'raw\/weather', 'low_temps.csv', true, ',', 'low_temps', 'date timestamp, temp float');\n<\/pre>\n
      Right away, we can see that will will want a gold view to join<\/b> the silver high and low tables into one logical dataset.  This is not an uncommon request.  I have a current client that has a old and new payroll system.  In the data lake, they want one view of the payroll register.  Thus, the view unions<\/b> several tables into one logical one.
      \n<\/p>\n
      Azure Data Lake Storage<\/h4>\n
      At this time, Microsoft Fabric does not support a on premise gateway<\/b>.  Use this link<\/a> to find out the major differences between Azure Synapse and Microsoft Fabric in regards to data factory.  In our proof of concept example, datasets will be dropped into a external storage account.<\/p>\n
      <\/p>\n
      To make life easy, the container and path will be the same between the source and destination.  The above image shows two temperature files and the below image shows one netflix file.<\/p>\n
      <\/p>\n
      To simply the example, we are going to use an account key to access the Azure Storage Account.<\/p>\n
      <\/p>\n
      For a production system, I will use a service principle since we can restrict access using both RBAC and ACL security.<\/p>\n
      <\/p>\n
      In the next section, we are going to start building our data pipeline.<\/p>\n
      Data Pipeline – Copy Activity<\/h4>\n
      Click the create icon on the left menu to start our Data Factory adventure.  Both Data Flow and Data Pipelines are available to the developer.  Please create a new<\/b> Data Pipeline.<\/p>\n
      <\/p>\n
      I am going to highlight the two activities that are key to our program.  The copy activity<\/b> allows a developer read data from the source and write data to the destination. The notebook activity<\/b> allows us to call a spark notebook that will tear down and rebuild the bronze and silver tables.<\/p>\n
      <\/p>\n
      Let’s work on the connection next.  In prior versions of data factory, the developer create a linked service<\/b> that contained the connection information and a dataset<\/b> that described the data.  For example, linked service named ls_asql_hris<\/b> has the connection to an Azure SQL database and a dataset named ds_asql_employee<\/b> describes the fields in the table.<\/p>\n
      These two objects are replaced by one connection object in Fabric.  The image below shows the support for many types of connections.  We want to setup on for Azure Data Lake Storage.<\/p>\n
      <\/p>\n
      Supply the fully qualified location to the storage as well as the account key.<\/p>\n
      <\/p>\n
      It is important to following your companies naming conventions for activities and add detailed comments when possible.
      \n The image below shows the copy activity we are about to configure.<\/p>\n
      <\/p>\n
      Since we are only copying the file from the source to the destination storage, please choose a binary copy<\/b>.  The image below shows the container name and file name are parameters to the source definition.  <\/p>\n
      <\/p>\n
      The same parameters are used for the destination with one addition.  We are going to add the current date formatted as “YYYYMMDD” as a subfolder.  That way, we can have multiple versions of the source files by a given day.<\/p>\n
      <\/p>\n
      The item() variable is exposed to the activity when a for-each activity<\/a> is used.  We will talk about working with meta data using the lookup activity next.<\/p>\n
      Data Pipeline – Lookup & ForEach – Activities<\/h4>\n
      The image below shows the top three activities used by our data pipeline.  The for-each activity has a two levels of nesting.  Please note I renamed the pipeline to pl-delimited-full-load<\/font> and gave it an appropriate description.  I defined a pipeline parameter named pipeline_id<\/font> which tells the pipeline which row of meta data to use.  <\/p>\n
      <\/p>\n
      The variable shown below was used in the destination section of the copy activity.<\/p>\n
      <\/p>\n
      Let’s review the details of the three topmost activities.  The image below shows the lookup activity<\/a> named lkp-meta-data<\/font> reads the meta_data table from the lake house and returns a JSON object. <\/p>\n
      <\/p>\n
      The set variable activity<\/a> named set-folder-var<\/font> converts the current data\/time into a pattern for the creation of the sub-folder in our destination storage.<\/p>\n
      <\/p>\n
      The for-each activity<\/a> traverses the array of meta data records.  The JSON output<\/b> from the lookup activity is feed to the for-each activity as input<\/b>.<\/p>\n
      <\/p>\n
      The simplest way to execute the copy and notebook activity for the correct pipeline id is to use a if condition activity<\/a>.<\/p>\n
      <\/p>\n
      The details of the condition are shown below.<\/p>\n
      <\/p>\n
      I usually do not repeat screen shots; However, this new feature in Data Factory is awesome!<\/b><\/p>\n
      <\/p>\n
      When testing each component of a pipeline, the developer might want to de-activate<\/a> or activate<\/a> a component.  I learn about this feature in the sequence container<\/a> when developing packages in SSIS 2005.  I am glad the feature made it to Data Factory.<\/p>\n
      To sum up this section, the lookup and for-each activity are really made for each other.  We could make this code more complex by figuring out if the JSON output from the lookup activity has an array element that matches our pipeline id.  However, all if condition tests are being done in parallel and it’s not worth the effort to save a few seconds.<\/p>\n
      Data Pipeline \u2013 Notebook Activity<\/h4>\n
      The notebook activity<\/a> just calls the Spark Notebook with the parameters we supply.  Please use a naming convention and enter detailed comments.<\/p>\n
      <\/p>\n
      On the settings page of the activity, we browse the workspace for the notebook we want to run.  In our case, the name of the notebook is lesson-04-full-load.  The parameters match one for one to the fields in the meta data table.<\/p>\n
      <\/p>\n
      Before we execute the finalized pipeline, let’s look at the revised Spark Notebook.<\/p>\n
      Spark Notebook<\/h4>\n
      At the top of the notebook we need a header to describe the program as well as a parameter cell.  The code below shows the parameter cell.<\/p>\n
      \nvar_path = 'raw\/netflix'\nvar_table = 'netflix'\nvar_delimiter = ','\nvar_header = 'true'\nvar_schema = ''\n<\/pre>\n
      The first step of the notebook is to remove the existing bronze and silver tables for a given parameterized table name.<\/p>\n
      \n#\n#  F1 - remove existing tables\n#\n\n# drop bronze table\nstmt = f'drop table if exists bronze_{var_table}_full;'\nret = spark.sql(stmt)\n\n# drop silver table\nstmt = f'drop table if exists silver_{var_table}_full;'\nret = spark.sql(stmt)\n<\/pre>\n
      The second step of the notebook is read the CSV file into a DataFrame and produce a temporary view as the final output.  Two tests are performed on the schema variable.  If the variable is None or a empty string after trimming, then we want to read infer the schema.<\/p>\n
      \n#\n#  F2 - load data frame\n#\n\n   \n# root path\npath = 'Files\/' + var_path\n\n# tmp table\ntable = 'tmp_' + var_table.strip() + '_full'\n\n# schema flag\nschema_flag = True\ntry:\n  \n  # none test\n  if var_schema is None:\n    schema_flag = False\n\n  # empty string \n  if not bool(var_schema.strip()):\n    schema_flag = False\n\nexcept:\n  pass\n\n\n# load all files w\/o schema\nif not schema_flag:\n    df = spark.read.format(\"csv\") \\\n        .option(\"header\",var_header) \\\n        .option(\"delimiter\", var_delimiter) \\\n        .option(\"recursiveFileLookup\", \"true\") \\\n        .load(path)\n\n# load all files w\/ schema\nelse:\n    df = spark.read.format(\"csv\") \\\n        .schema(var_schema) \\\n        .option(\"header\",var_header) \\\n        .option(\"delimiter\", var_delimiter) \\\n        .option(\"recursiveFileLookup\", \"true\") \\\n        .load(path)\n\n# convert to view\ndf.createOrReplaceTempView(table)\n\n# debugging\nprint(f\"data lake path - {path}\")\nprint(f\"temporary view - {table}\")\n<\/pre>\n
      The third step in the program creates the bronze table while adding a load date, a folder name and a file name.  The split_string function is called with parameters specific to our directory structure.<\/p>\n
      \n#\n#  F3 - create bronze table (all files)\n#\n  \n# spark sql - assume 1 level nesting on dir\nstmt = f\"\"\"\n  create table bronze_{var_table}_full as\n  select \n    *, \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_{var_table}_full \n\"\"\"\n\n# create table\nret = spark.sql(stmt)\n\n# debugging\nprint(f\"execute spark sql - \\n {stmt}\")\n<\/pre>\n
      The fourth step creates the silver table by only showing the data from the newest sub-directory by date string.<\/p>\n
      \n#\n#  F4 - create silver table (lastest file)\n#\n  \n# spark sql\nstmt = f\"\"\"\n  create table silver_{var_table}_full as\n  with cte_{var_table} as\n  (\n    select * \n    from bronze_{var_table}_full as l\n    where l._folder_name = (select max(_folder_name) from bronze_{var_table}_full)\n  )\n  select \n    *\n  from \n    cte_{var_table}\n\"\"\"\n\n# create table\nret = spark.sql(stmt)\n\n# debugging\nprint(f\"execute spark sql - \\n {stmt}\")\n<\/pre>\n
      If this was a production system, I would be logging the record counts into an audit table.  Here is code to print record counts for the bronze table.<\/p>\n
      \n#\n#  Grab bronze count\n#\n\ntry:\n    sql_stmt = f\"select count(*) from bronze_{var_table}_full\"\n    bronze_recs = spark.sql(sql_stmt).first()[0]\nexcept:\n    bronze_recs = 0\n\n# show values\nprint(f\"The bronze_{var_table}_full record count is {bronze_recs}\")  \n\n<\/pre>\n
      Of course, we want to show record counts for the silver table.<\/p>\n
      \n#\n#  Grab silver count\n#\n\ntry:\n    sql_stmt = f\"select count(*) from silver_{var_table}_full\"\n    silver_recs = spark.sql(sql_stmt).first()[0]\nexcept:\n    silver_recs = 0\n\n# show values\nprint(f\"The silver_{var_table}_full record count is {silver_recs}\")  \n<\/pre>\n
      Now that will have a complete Data Pipeline – copy data into lake and Spark Notebook – transform files into tables, lets execute the pipeline three times with each pipeline id to load the CSV files into our one lake (lake house).<\/p>\n
      Testing<\/h4>\n
      Please set the pipeline variable named pipeline id to one.  Then trigger the pipeline.  The output below shows a successful execution of the code.<\/p>\n
      <\/p>\n
      If will click on the output of the Spark Notebook, we get a link to the notebook snapshot. <\/p>\n
      <\/p>\n
      This snapshot was taken from the high temperature execution.  We can tell by the debugging statements and record counts.<\/p>\n
      <\/p>\n
      At the end of the first execution, we have create a bronze and silver table for the Netflix titles.  The image below shows a grid with the data.<\/p>\n
      <\/p>\n
      At this time, please repeat the execution for each source.  This can be done by incrementing the pipeline id and executing the pipeline.<\/p>\n
      <\/p>\n
      While creating a meta data driven design takes some time, it is well worth the investment.  Now, I can load hundreds of files into the lake by adding meta data to the table and scheduling the pipeline with the correct id.<\/p>\n
      Summary<\/h4>\n
      Overall, I really do like the one stop shop<\/b> idea of Fabric.  Every you need is in one place with the Microsoft Fabric Service.  Like any development team, Microsoft had to push the product out the door with a few rough edges. <\/p>\n
      The image below shows the three items that might have larger customers wait on adoption.<\/p>\n
      <\/p>\n
      First<\/font>, many companies still have a good percentage of their data on premises.  Without a data gateway, older technologies such as SSIS will have to be used to land the data into a storage container.  <\/p>\n
      Second<\/font>, some companies like hospitals and financial institutions regard their data as highly sanative. The use of a virtual private network and endpoints has secured Azure components in the past.  This support is not available in Fabric today.<\/p>\n
      Third<\/font>, most companies have at least two environments:  Development and Production.  How do you move code from one environment to another?  How do you keep track of versions?  Spark Notebooks are easy since you can just download and upload files and manually check the code into git.  <\/p>\n
      Data Factory is more complicated.  It always has been a JSON export file that had many inter-dependencies.  But some of that has gone away.  A linked service and dataset has been replaced with a connection which is part of the pipeline.  Therefore, it might be possible to download and upload pipelines in the future?  For now, we can move code by create a pipeline with the same name as the demo – pl-delimited-full-load<\/font>.  Then hit the {} button to modify the JSON with the supplied definition in the zip file. <\/p>\n
      In short, the data gateway, private end points, and continuous integration \/ continuous deployment features are required to make this product shine. Enclosed is the zip file<\/a> with the CSV data files, Data Pipeline in JSON format and Spark notebooks.<\/p>\n","protected":false},"excerpt":{"rendered":"
      Metadata Driven Pipelines What is a metadata driven pipeline? Wikipedia defines metadata as “data that provides information about other data”. As a developer, we can create a non parameterized pipeline and\/or notebook to solve a business problem. However, if we have to solve the same problem a hundred times, the amount of code can get unwieldly. A better way to solve this problem is to store metadata in the delta lake. This data will drive how the Azure Data Factory and Spark Notebooks execute. Business Problem Our manager has asked…<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1132],"tags":[1136],"class_list":["post-9447","post","type-post","status-publish","format-standard","hentry","category-microsoft-fabric","tag-john-f-miner-iii-microsoft-fabric-data-engineering-apache-spark-data-factory-full-load-code-meta-data-driven-design-learn"],"_links":{"self":[{"href":"https:\/\/craftydba.com\/index.php?rest_route=\/wp\/v2\/posts\/9447","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/craftydba.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/craftydba.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/craftydba.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/craftydba.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=9447"}],"version-history":[{"count":212,"href":"https:\/\/craftydba.com\/index.php?rest_route=\/wp\/v2\/posts\/9447\/revisions"}],"predecessor-version":[{"id":9796,"href":"https:\/\/craftydba.com\/index.php?rest_route=\/wp\/v2\/posts\/9447\/revisions\/9796"}],"wp:attachment":[{"href":"https:\/\/craftydba.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=9447"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/craftydba.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=9447"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/craftydba.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=9447"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}