{"id":1023,"date":"2021-01-27T01:27:40","date_gmt":"2021-01-27T01:27:40","guid":{"rendered":"http:\/\/jackofalltradesmasterofsome.com\/blog\/?p=1023"},"modified":"2021-02-05T14:35:12","modified_gmt":"2021-02-05T14:35:12","slug":"streaming-etl-with-azure-data-factory-and-create-a-parameter-driver-pipeline","status":"publish","type":"post","link":"https:\/\/jackofalltradesmasterofsome.com\/blog\/2021\/01\/27\/streaming-etl-with-azure-data-factory-and-create-a-parameter-driver-pipeline\/","title":{"rendered":"Streaming ETL with Azure Data Factory and CDC – Create a Parameter Driver Pipeline"},"content":{"rendered":"\n

In this series we look at building a Streaming ETL with Azure Data Factory and CDC – Create a Parameter Driver Pipeline. This is Part 6, The rest of the series is below.<\/p>\n\n\n\n

  1. Enabling CDC<\/a><\/li>
  2. Setting up Audit Tables<\/a><\/li>
  3. Provisioning Azure Data Factory<\/a><\/li>
  4. Provisioning Azure Blog Storage<\/a><\/li>
  5. Create Data Source Connection in ADF<\/a><\/li>
  6. Create Incremental Pipeline in ADF<\/a><\/li>
  7. Create a Parameter Driven Pipeline<\/a><\/li>
  8. Create a Rolling Trigger<\/a><\/li><\/ol>\n\n\n\n

    This series uses the Adventureworks database. For more information on how to get that set up see my Youtube video for Downloading <\/a>and Restoring<\/a> the database. <\/p>\n\n\n\n

    The previous step will\npull all the changes in the CDC table, but we do not want to do this all the\ntime. So let\u2019s look at creating a rolling window for the CDC ETL.<\/p>\n\n\n\n

    1. Navigate to the parameters section and create a new parameter. Add two paramenters \u201ctriggerStartTime\u201d and triggerEndTime\u201d and set them to yesterday and todays date in the format \u201c2020-01-07 12:00:00:000\u201d<\/li><\/ol>\n\n\n\n
      \"\"<\/figure>\n\n\n\n
      \"\"<\/figure>\n\n\n\n
      • On the Lookup Activity, update the query in the settings to the following to use the new variables. SQL Agent must be running for this step the parameters must be valid dates.<\/li><\/ul>\n\n\n\n

        @concat(‘DECLARE @begin_time\ndatetime, @end_time datetime, @from_lsn binary(10), @to_lsn binary(10); <\/p>\n\n\n\n

        SET @begin_time =\n”’,pipeline().parameters.triggerStartTime,”’;<\/p>\n\n\n\n

        SET @end_time =\n”’,pipeline().parameters.triggerEndTime,”’;<\/p>\n\n\n\n

        SET @from_lsn =\nsys.fn_cdc_map_time_to_lsn(”smallest greater than or equal”, @begin_time);<\/p>\n\n\n\n

        SET @to_lsn = sys.fn_cdc_map_time_to_lsn(”largest\nless than”, @end_time);<\/p>\n\n\n\n

        SELECT count(1) changecount FROM cdc.fn_cdc_get_all_changes_dbo_DimProduct (@from_lsn, @to_lsn, ”all”)’)<\/p>\n\n\n\n

        \"\"<\/figure>\n\n\n\n
        • Navigate back to the \u201cTrue\u201d condition and paste\nthe following query in to track the changes with the variables as well<\/li><\/ul>\n\n\n\n

          @concat(‘DECLARE @begin_time\ndatetime, @end_time datetime, @from_lsn binary(10), @to_lsn binary(10); <\/p>\n\n\n\n

          SET @begin_time =\n”’,pipeline().parameters.triggerStartTime,”’;<\/p>\n\n\n\n

          SET @end_time =\n”’,pipeline().parameters.triggerEndTime,”’;<\/p>\n\n\n\n

          SET @from_lsn =\nsys.fn_cdc_map_time_to_lsn(”smallest greater than or equal”, @begin_time);<\/p>\n\n\n\n

          SET @to_lsn =\nsys.fn_cdc_map_time_to_lsn(”largest less than”, @end_time);<\/p>\n\n\n\n

          SELECT * FROM cdc.fn_cdc_get_all_changes_dbo_DimProduct(@from_lsn, @to_lsn, ”all”)’)<\/p>\n\n\n\n

          \"\"<\/figure>\n\n\n\n
          • Edit the Sink tab in the true statement and click on parameters.<\/li><\/ul>\n\n\n\n
            \"\"<\/figure>\n\n\n\n
            \"\"<\/figure>\n\n\n\n
            • Add a new parameter called triggerStart<\/li><\/ul>\n\n\n\n
              \"\"<\/figure>\n\n\n\n
              • Head back to the Connections Tab for the dataset where we will be adding dynamic content for the directory and file.<\/li><\/ul>\n\n\n\n
                \"\"<\/figure>\n\n\n\n
                • Add the following for the directory and file\nsections.<\/li><\/ul>\n\n\n\n

                  Directory<\/strong><\/p>\n\n\n\n

                  @concat(‘dimProduct\/incremental\/’,formatDateTime(dataset().triggerStart,’yyyy\/MM\/dd’)) <\/p>\n\n\n\n

                  \"\"<\/figure>\n\n\n\n

                  File<\/strong><\/p>\n\n\n\n

                  @concat(formatDateTime(dataset().triggerStart,’yyyyMMddHHmmssf<\/p>\n\n\n\n

                  ff’),’.csv’)<\/p>\n\n\n\n

                  \"\"<\/figure>\n\n\n\n
                  • Navigate back to the Sink in the Copy and expand dataset properties. Add the dynamic content for the new parameter.<\/li><\/ul>\n\n\n\n
                    \"\"<\/figure>\n\n\n\n
                    1. You can now trigger your run and see the new files landing in the datalake.<\/li><\/ol>\n\n\n\n

                      Streaming ETL with Azure Data Factory and CDC – Create a Parameter Driver Pipeline <\/p>\n","protected":false},"excerpt":{"rendered":"

                      In this series we look at building a Streaming ETL with Azure Data Factory and CDC – Create a Parameter Driver Pipeline. This is Part 6, The rest of the series is below. Enabling CDC Setting up Audit Tables Provisioning Azure Data Factory Provisioning Azure Blog Storage Create Data Source Connection in ADF Create Incremental […]<\/p>\n","protected":false},"author":1,"featured_media":822,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[4,28,27,11,1],"tags":[104],"class_list":["post-1023","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-analytics","category-azure","category-big-data","category-data-warehouse","category-uncategorized","tag-azure-data-factory"],"_links":{"self":[{"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/posts\/1023","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/comments?post=1023"}],"version-history":[{"count":2,"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/posts\/1023\/revisions"}],"predecessor-version":[{"id":1079,"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/posts\/1023\/revisions\/1079"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/media\/822"}],"wp:attachment":[{"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/media?parent=1023"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/categories?post=1023"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/jackofalltradesmasterofsome.com\/blog\/wp-json\/wp\/v2\/tags?post=1023"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}