Rozgałęzianie i łączenie działań w potoku Azure Data Factory przy użyciu Azure Portal

DOTYCZY: Azure Data Factory Azure Synapse Analytics

W tym samouczku pokazano, jak utworzyć potok usługi Data Factory przedstawiający niektóre funkcje przepływu sterowania. Ten potok tworzy prostą kopię z kontenera w usłudze Azure Blob Storage w innym kontenerze na tym samym koncie magazynu. Jeśli działanie kopiowania zakończy się powodzeniem, potok wysyła szczegóły zakończonej pomyślnie operacji kopiowania (takie jak ilość zapisanych danych) w wiadomości e-mail z informacją o powodzeniu. W przypadku niepowodzenia działania kopiowania potok wysyła szczegóły błędu kopiowania (np. komunikat o błędzie) w wiadomości e-mail z informacją o niepowodzeniu. W samouczku pokazano, jak przekazać parametry.

Ogólne omówienie scenariusza: Diagram shows Azure Blob Storage, which is the target of a copy, which, on success, sends an email with details or, on failure, sends an email with error details.

Ten samouczek obejmuje następujące procedury:

  • Tworzenie fabryki danych.
  • Tworzenie połączonej usługi Azure Storage
  • Tworzenie zestawu danych obiektów blob platformy Azure
  • Tworzenie potoku zawierającego działanie kopiowania i działanie internetowe
  • Wysyłanie danych wyjściowych działań do kolejnych działań
  • Korzystanie z przekazywania parametrów i ze zmiennych systemowych
  • Uruchamianie potoku
  • Monitorowanie uruchomień działań i potoku

W tym samouczku jest używana witryna Azure Portal. Aby uzyskać informacje o innych mechanizmach interakcji z usługą Azure Data Factory, przejdź do przewodników Szybki start w spisie treści.

Wymagania wstępne

  • Subskrypcja platformy Azure. Jeśli nie masz subskrypcji platformy Azure, przed rozpoczęciem utwórz bezpłatne konto.
  • Konto usługi Azure Storage. Magazyn obiektów blob jest używany jako źródłowy magazyn danych. Jeśli nie masz konta usługi Azure Storage, utwórz je, wykonując czynności przedstawione w artykule Tworzenie konta magazynu.
  • Usługa Azure SQL Database. Baza danych jest używana jako magazyn danych ujścia. Jeśli nie masz bazy danych w Azure SQL Database, zobacz artykuł Tworzenie bazy danych w Azure SQL Database, aby uzyskać instrukcje tworzenia bazy danych.

Tworzenie tabeli obiektów blob

  1. Uruchom program Notatnik. Skopiuj poniższy tekst i zapisz go na dysku jako plik input.txt.

    John,Doe
    Jane,Doe
    
  2. Użyj narzędzia takiego jak Eksplorator usługi Azure Storage, aby wykonać następujące czynności:

    1. Utwórz kontener adfv2branch.
    2. Utwórz folder input w kontenerze adfv2branch.
    3. Przekaż plik input.txt do kontenera.

Tworzenie punktów końcowych przepływu pracy poczty e-mail

Aby wyzwolić wysyłanie wiadomości e-mail z potoku, zdefiniuj przepływ pracy przy użyciu usługi Logic Apps. Aby uzyskać szczegółowe informacje na temat tworzenia przepływu pracy aplikacji logiki, zobacz Jak utworzyć aplikację logiki.

Przepływ pracy wiadomości e-mail z informacją o powodzeniu

Utwórz przepływ pracy aplikacji logiki o nazwie CopySuccessEmail. Zdefiniuj wyzwalacz przepływu pracy jako When an HTTP request is received i dodaj akcję Office 365 Outlook – Send an email.

Success email workflow

W wyzwalaczu żądania wypełnij pole Request Body JSON Schema przy użyciu następującego kodu JSON:

{
    "properties": {
        "dataFactoryName": {
            "type": "string"
        },
        "message": {
            "type": "string"
        },
        "pipelineName": {
            "type": "string"
        },
        "receiver": {
            "type": "string"
        }
    },
    "type": "object"
}

Żądanie w projektancie aplikacji logiki powinno wyglądać jak na poniższym obrazie:

Logic App designer - request

W przypadku akcji Wyślij wiadomość e-mail dostosuj formatowanie wiadomości e-mail przy użyciu właściwości przekazywanych w schemacie JSON w treści żądania. Oto przykład:

Logic App designer - send email action

Zapisz przepływ pracy. Zapisz adres URL żądania HTTP POST dla przepływu pracy wiadomości e-mail z informacją o powodzeniu:

//Success Request Url
https://prodxxx.eastus.logic.azure.com:443/workflows/000000/triggers/manual/paths/invoke?api-version=2016-10-01&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=000000

Przepływ pracy wiadomości e-mail z informacją o niepowodzeniu

Wykonaj te same kroki, aby utworzyć inny przepływ pracy usługi Logic Apps o nazwie CopyFailEmail. W wyzwalaczu żądania element Request Body JSON schema jest taki sam. Zmień format wiadomości e-mail, na przykład element Subject, aby przekształcić wiadomość e-mail w wiadomość z informacją o niepowodzeniu. Oto przykład:

Logic App designer - fail email workflow

Zapisz przepływ pracy. Zapisz adres URL żądania HTTP POST dla przepływu pracy wiadomości e-mail z informacją o niepowodzeniu:

//Fail Request Url
https://prodxxx.eastus.logic.azure.com:443/workflows/000000/triggers/manual/paths/invoke?api-version=2016-10-01&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=000000

Teraz powinny być zapisane dwa adresy URL przepływów pracy:

//Success Request Url
https://prodxxx.eastus.logic.azure.com:443/workflows/000000/triggers/manual/paths/invoke?api-version=2016-10-01&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=000000

//Fail Request Url
https://prodxxx.eastus.logic.azure.com:443/workflows/000000/triggers/manual/paths/invoke?api-version=2016-10-01&sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=000000

Tworzenie fabryki danych

  1. Uruchom przeglądarkę internetową Microsoft Edge lub Google Chrome. Obecnie interfejs użytkownika usługi Data Factory jest obsługiwany tylko przez przeglądarki internetowe Microsoft Edge i Google Chrome.

  2. W menu po lewej stronie wybierz pozycję Utwórz zasóbData>+ AnalyticsData>Factory:

    Data Factory selection in the "New" pane

  3. Na stronie Nowa fabryka danych wprowadź wartość ADFTutorialDataFactory w polu Nazwa.

    New data factory page

    Nazwa fabryki danych platformy Azure musi być globalnie unikatowa. Jeśli wystąpi poniższy błąd, zmień nazwę fabryki danych (np. twojanazwaADFTutorialDataFactory) i spróbuj utworzyć ją ponownie. Artykuł Data Factory — Naming Rules (Usługa Data Factory — reguły nazewnictwa) zawiera reguły nazewnictwa artefaktów usługi Data Factory.

    Nazwa fabryki danych "ADFTutorialDataFactory" jest niedostępna.

  4. Wybierz subskrypcję Azure, w której chcesz utworzyć fabrykę danych.

  5. Dla opcji Grupa zasobów wykonaj jedną z następujących czynności:

    • Wybierz pozycję Użyj istniejącej, a następnie wybierz istniejącą grupę zasobów z listy rozwijanej.

    • Wybierz pozycję Utwórz nową, a następnie wprowadź nazwę grupy zasobów.

      Informacje na temat grup zasobów znajdują się w artykule Using resource groups to manage your Azure resources (Używanie grup zasobów do zarządzania zasobami platformy Azure).

  6. Wybierz opcję V2 w obszarze Wersja.

  7. Na liście lokalizacja wybierz lokalizację fabryki danych. Na liście rozwijanej są wyświetlane tylko obsługiwane lokalizacje. Magazyny danych (Azure Storage, Azure SQL Database itp.) i jednostki obliczeniowe (HDInsight itp.) używane przez fabrykę danych mogą mieścić się w innych regionach.

  8. Wybierz opcję Przypnij do pulpitu nawigacyjnego.

  9. Kliknij pozycję Utwórz.

  10. Na pulpicie nawigacyjnym zobaczysz następujący kafelek ze stanem: Wdrażanie fabryki danych.

    deploying data factory tile

  11. Po zakończeniu tworzenia zostanie wyświetlona strona Fabryka danych, jak pokazano na poniższej ilustracji.

    Data factory home page

  12. Kliknij kafelek Author Monitor (Author & Monitor), aby uruchomić interfejs użytkownika Azure Data Factory na osobnej karcie.

Tworzenie potoku

W tym kroku jest tworzony potok z jednym działaniem kopiowania i dwoma działaniami internetowymi. Do utworzenia potoku zostaną użyte następujące funkcje:

  • Parametry dla potoku, do których uzyskują dostęp zestawy danych.
  • Działania internetowe w celu wywołania przepływów pracy aplikacji logiki wysyłających wiadomości e-mail z informacją o powodzeniu/niepowodzeniu.
  • Łączenie jednego działania z innym (w przypadku powodzenia i niepowodzenia)
  • Używanie danych wyjściowe działania jako danych wejściowych kolejnego działania
  1. Na stronie głównej interfejsu użytkownika usługi Data Factory kliknij kafelek Orkiestracja .

    Screenshot that shows the ADF home page.

  2. W oknie właściwości potoku przejdź do karty Parametry i użyj przycisku Nowy, aby dodać następujące trzy parametry typu String: sourceBlobContainer, sinkBlobContainer oraz receiver.

    • sourceBlobContainer — parametr w potoku używany przez zestaw danych obiektu blob źródła.
    • sinkBlobContainer — parametr w potoku używany przez zestaw danych obiektu blob ujścia.
    • odbiornik — ten parametr jest używany przez dwie działania sieci Web w potoku, które wysyłają wiadomości e-mail o powodzeniu lub niepowodzeniu do odbiorcy, którego adres e-mail jest określony przez ten parametr.

    New pipeline menu

  3. W przyborniku Działania rozwiń pozycję Przepływ danych, a następnie przeciągnij działanie Kopiowanie i upuść je na powierzchni projektanta potoku.

    Drag-drop copy activity

  4. W oknie Właściwości dla działania Kopiowanie znajdującego się na dole przejdź do karty Źródło, a następnie kliknij pozycję + Nowy. W tym kroku utworzysz zestaw danych źródłowych dla działania kopiowania.

    Screenshot that shows how to create a source dataset for teh copy activity.

  5. W oknie Nowy zestaw danych wybierz pozycję Azure Blob Storage i kliknij przycisk Zakończ.

    Select Azure Blob Storage

  6. Zostanie wyświetlona nowa karta zatytułowana AzureBlob1. Zmień nazwę zestawu danych na SourceBlobDataset.

    Dataset general settings

  7. Przejdź do karty Połączenie w oknie Właściwości, a następnie kliknij przycisk Nowy dla pozycji Połączona usługa. W tym kroku utworzysz połączoną usługę służącą do łączenia konta usługi Azure Storage z fabryką danych.

    Dataset connection - new linked service

  8. W oknie Nowa połączona usługa wykonaj następujące czynności:

    1. Wprowadź wartość AzureStorageLinkedService w polu Nazwa.
    2. Wybierz swoje konto usługi Azure Storage w polu Nazwa konta magazynu.
    3. Kliknij pozycję Zapisz.

    New Azure Storage linked service

  9. Wprowadź wartość @pipeline().parameters.sourceBlobContainer jako folder oraz emp.txt jako nazwę pliku. Parametr potoku sourceBlobContainer umożliwia ustawienie ścieżki folderu dla zestawu danych.

    Source dataset settings

  10. Przejdź do karty potoku lub kliknij potok w widoku drzewa. Upewnij się, że dla ustawienia Zestaw danych źródłowych wybrano wartość SourceBlobDataset.

Source dataset

  1. W oknie Właściwości przejdź do karty Ujście, a następnie kliknij pozycję + Nowy dla elementu Zestaw danych ujścia. W tym kroku utworzysz zestaw danych ujścia dla działania kopiowania w podobny sposób do tworzenia zestawu danych źródłowych.

    New sink dataset button

  2. W oknie Nowy zestaw danych wybierz pozycję Azure Blob Storage i kliknij przycisk Zakończ.

  3. Na stronie ustawień Ogólne dla zestawu danych wprowadź wartość SinkBlobDataset w polu Nazwa.

  4. Przejdź do karty Połączenie i wykonaj następujące czynności:

    1. Wybierz pozycję AzureStorageLinkedService dla elementu LinkedService.

    2. Wprowadź wartość @pipeline().parameters.sinkBlobContainer jako folder.

    3. Wprowadź wartość @CONCAT(pipeline().RunId, '.txt') jako nazwę pliku. Wyrażenie wykorzystuje identyfikator bieżącego uruchomienia potoku dla nazwy pliku. Aby uzyskać listę obsługiwanych zmiennych systemowych i wyrażeń, zobacz Zmienne systemowe i Język wyrażeń.

      Sink dataset settings

  5. Przejdź do karty potoku u góry. W przyborniku Działania rozwiń pozycję Ogólne, a następnie przeciągnij działanie Internet i upuść je na powierzchni projektanta potoku. Ustaw nazwę działania na wartość SendSuccessEmailActivity. Działanie internetowe umożliwia wywołanie dowolnego punktu końcowego REST. Aby uzyskać więcej informacji na temat działania, zobacz Działanie internetowe. Ten potok używa działania internetowego w celu wywołania przepływu pracy poczty e-mail usługi Logic Apps.

    Drag-drop first Web activity

  6. Z karty Ogólne przejdź do karty Ustawienia i wykonaj następujące czynności:

    1. W pozycji Adres URL określ adres URL dla przepływu pracy aplikacji logiki, który wysyła wiadomość e-mail z informacją o powodzeniu.

    2. Wybierz wartość POST dla pozycji Metoda.

    3. Kliknij link + Dodaj nagłówek w sekcji Nagłówki.

    4. Dodaj nagłówek Content-Type i ustaw go na wartość application/json.

    5. Podaj następujący kod JSON w pozycji Treść.

      {
          "message": "@{activity('Copy1').output.dataWritten}",
          "dataFactoryName": "@{pipeline().DataFactory}",
          "pipelineName": "@{pipeline().Pipeline}",
          "receiver": "@pipeline().parameters.receiver"
      }
      

      Treść wiadomości zawiera następujące właściwości:

      • Wiadomość — przekazywanie wartości elementu @{activity('Copy1').output.dataWritten. Uzyskuje dostęp do właściwości poprzedniego działania kopiowania i przekazuje wartość elementu dataWritten. W przypadku wiadomości dotyczącej niepowodzenia przekaż dane wyjściowe błędu zamiast elementu @{activity('CopyBlobtoBlob').error.message.

      • Nazwa fabryki danych — przekazywanie wartości elementu @{pipeline().DataFactory}. Jest to zmienna systemowa, która umożliwia dostęp do odpowiedniej nazwy fabryki danych. Lista zmiennych systemowych jest dostępna w artykule Zmienne systemowe.

      • Nazwa potoku — przekazywanie wartości elementu @{pipeline().Pipeline}. Jest to również zmienna systemowa, która umożliwia dostęp do odpowiedniej nazwy potoku.

      • Receiver — przekazywanie wartości "@pipeline().parameters.receiver"). Uzyskuje dostęp do parametrów potoku.

        Settings for the first Web activity

  7. Połącz działanie Kopiowanie z działaniem Internet, przeciągając zielony przycisk znajdujący się obok działania kopiowania i upuszczając go na działaniu internetowym.

    Connect Copy activity with the first Web activity

  8. Przeciągnij kolejne działanie Internet z przybornika Działania i upuść je na powierzchni projektanta potoku, a następnie ustaw nazwę na wartość SendFailureEmailActivity.

    Name of the second Web activity

  9. Przejdź do karty Ustawienia i wykonaj następujące czynności:

    1. W pozycji Adres URL określ adres URL dla przepływu pracy aplikacji logiki, który wysyła wiadomość e-mail z informacją o niepowodzeniu.

    2. Wybierz wartość POST dla pozycji Metoda.

    3. Kliknij link + Dodaj nagłówek w sekcji Nagłówki.

    4. Dodaj nagłówek Content-Type i ustaw go na wartość application/json.

    5. Podaj następujący kod JSON w pozycji Treść.

      {
          "message": "@{activity('Copy1').error.message}",
          "dataFactoryName": "@{pipeline().DataFactory}",
          "pipelineName": "@{pipeline().Pipeline}",
          "receiver": "@pipeline().parameters.receiver"
      }
      

      Settings for the second Web activity

  10. Wybierz pozycję Kopiuj działanie w projektancie potoku, a następnie kliknij przycisk +-> i wybierz pozycję Błąd.

    Screenshot that shows how to select Error on the Copy activity in the pipeline designer.

  11. Przeciągnij czerwony przycisk znajdujący się obok działania kopiowania do drugiego działania internetowego SendFailureEmailActivity. Działania można przenosić, tak aby potok wyglądał jak na poniższym obrazie:

    Full pipeline with all activities

  12. Aby zweryfikować potok, kliknij przycisk Weryfikuj na pasku narzędzi. Zamknij okno Dane wyjściowe weryfikacji potoku, klikając przycisk >>.

    Validate pipeline

  13. Aby opublikować jednostki (zestawy danych, potoki itp.) w usłudze Data Factory, kliknij przycisk Opublikuj wszystko. Poczekaj na wyświetlenie komunikatu Pomyślnie opublikowano.

    Publish

Wyzwalanie pomyślnego uruchomienia potoku

  1. Aby wyzwolić uruchomienie potoku, kliknij pozycję Wyzwól na pasku narzędzi, a następnie kliknij polecenie Wyzwól teraz.

    Trigger a pipeline run

  2. W oknie Uruchomienie potoku wykonaj następujące czynności:

    1. Wprowadź wartość adftutorial/adfv2branch/input dla parametru sourceBlobContainer.

    2. Wprowadź wartość adftutorial/adfv2branch/output dla parametru sinkBlobContainer.

    3. Wprowadź adres e-mail dla parametru receiver.

    4. Kliknij przycisk Zakończ

      Pipeline run parameters

Monitorowanie pomyślnego uruchomienia potoku

  1. Aby monitorować uruchomienie potoku, przejdź do karty Monitorowanie po lewej stronie. Zostanie wyświetlone uruchomienie potoku, które zostało wyzwolone ręcznie przez Ciebie. Kliknij przycisk Odśwież, aby odświeżyć listę.

    Successful pipeline run

  2. Aby wyświetlić uruchomienia działań skojarzone z tym uruchomieniem potoku, kliknij pierwszy link w kolumnie Akcje. Do poprzedniego widoku można wrócić, klikając pozycję Potoki u góry. Kliknij przycisk Odśwież, aby odświeżyć listę.

    Screenshot that shows how to view the list of activity runs.

Wyzwalanie niepomyślnego uruchomienia potoku

  1. Przejdź do karty Edycja po lewej stronie.

  2. Aby wyzwolić uruchomienie potoku, kliknij pozycję Wyzwól na pasku narzędzi, a następnie kliknij polecenie Wyzwól teraz.

  3. W oknie Uruchomienie potoku wykonaj następujące czynności:

    1. Wprowadź wartość adftutorial/dummy/input dla parametru sourceBlobContainer. Upewnij się, że folder „dummy” nie istnieje w kontenerze adftutorial.
    2. Wprowadź wartość adftutorial/dummy/output dla parametru sinkBlobContainer.
    3. Wprowadź adres e-mail dla parametru receiver.
    4. Kliknij przycisk Finish (Zakończ).

Monitorowanie zakończonego niepowodzeniem uruchomienia potoku

  1. Aby monitorować uruchomienie potoku, przejdź do karty Monitorowanie po lewej stronie. Zostanie wyświetlone uruchomienie potoku, które zostało wyzwolone ręcznie przez Ciebie. Kliknij przycisk Odśwież, aby odświeżyć listę.

    Failure pipeline run

  2. Kliknij link Błąd dla uruchomienia potoku, aby wyświetlić szczegóły dotyczące błędu.

    Pipeline error

  3. Aby wyświetlić uruchomienia działań skojarzone z tym uruchomieniem potoku, kliknij pierwszy link w kolumnie Akcje. Kliknij przycisk Odśwież, aby odświeżyć listę. Zwróć uwagę, że działanie kopiowania w potoku nie powiodło się. Działanie internetowe powiodło się i do określonego odbiorcy została wysłana wiadomość o niepowodzeniu.

    Activity runs

  4. Kliknij link Błąd w kolumnie Akcje, aby wyświetlić szczegóły dotyczące błędu.

    Activity run error

Następne kroki

W ramach tego samouczka wykonano następujące procedury:

  • Tworzenie fabryki danych.
  • Tworzenie połączonej usługi Azure Storage
  • Tworzenie zestawu danych obiektów blob platformy Azure
  • Tworzenie potoku zawierającego działanie kopiowania i działanie internetowe
  • Wysyłanie danych wyjściowych działań do kolejnych działań
  • Korzystanie z przekazywania parametrów i ze zmiennych systemowych
  • Uruchamianie potoku
  • Monitorowanie uruchomień działań i potoku

Teraz możesz przejść do sekcji Pojęcia, aby uzyskać więcej informacji na temat usługi Azure Data Factory.