Zelfstudie: Het Data Lake Capture-patroon implementeren om een Databricks Delta-tabel bij te werken

In deze zelfstudie leert u hoe u gebeurtenissen in een opslagaccount met een hiërarchische naamruimte verwerkt.

Je bouwt een kleine oplossing waarmee je een Databricks Delta-tabel kunt vullen door een CSV-bestand (comma-separated values) te uploaden dat een verkooporder beschrijft. Je bouwt deze oplossing door een Event Grid-abonnement, een Azure-functie en een taak te koppelen in Azure Databricks.

In deze handleiding leert u:

  • Maak een Event Grid-abonnement die een Azure-functie aanroept.
  • Maak een Azure-functie die een melding ontvangt van een gebeurtenis en vervolgens de taak uitvoert in Azure Databricks.
  • Maak een Databricks-taak waarmee een klantorder wordt ingevoegd in een Databricks Delta-tabel in het opslagaccount.

Je bouwt deze oplossing in omgekeerde volgorde, te beginnen met de Azure Databricks workspace.

Vereisten

Een verkooporder maken

Maak eerst een CSV-bestand aan dat een verkooporder beschrijft, en upload dat bestand vervolgens naar het opslagaccount. Later gebruik je de gegevens uit dit bestand om de eerste rij in je Databricks Delta-tabel te vullen.

  1. Ga naar uw nieuwe opslagaccount in de Azure-portal.

  2. Selecteer Storage browser>Blob containers>Voeg container toe en maak een nieuwe container aan met de naam data.

    Screenshot van het aanmaken van een container in de Azure Storage-browser.

  3. Maak in de gegevenscontainer een map met de naam Invoer.

  4. Plak de volgende tekst in een teksteditor.

    InvoiceNo,StockCode,Description,Quantity,InvoiceDate,UnitPrice,CustomerID,Country
536365,85123A,WHITE HANGING HEART T-LIGHT HOLDER,6,12/1/2010 8:26,2.55,17850,United Kingdom

  5. Sla dit bestand op op je lokale computer en noem het data.csv.

  6. Upload dit bestand in de Storage browser naar de invoermap .

Een taak maken in Azure Databricks

In deze sectie voer je de volgende taken uit:

  • Een Azure Databricks-werkruimte maken.
  • Maak een notebook.
  • Een Databricks Delta-tabel maken en invullen.
  • Code toevoegen waarmee rijen in de Databricks Delta-tabel worden ingevoegd.
  • Maak een job.

Een Azure Databricks-werkruimte maken

In deze sectie maak je een Azure Databricks-werkruimte aan door gebruik te maken van het Azure-portaal.

  1. Een Azure Databricks-werkruimte maken. Noem de werkruimte.contoso-orders Zie Een Azure Databricks-werkruimte maken.

  2. Een cluster maken. Geef het cluster customer-order-clustereen naam. Zie Een cluster maken.

  3. Maak een notebook. Geef het notebook configure-customer-table een naam en kies Python als de standaardtaal van het notebook. Zie Een notitieblok maken.

Een Databricks Delta-tabel maken en invullen

  1. Kopieer en plak het volgende codeblok in de eerste cel van de notebook die u hebt gemaakt, maar voer deze code nog niet uit.

    Vervang de appId, password, en tenant tijdelijke waarden in dit codeblok door de waarden die je hebt verzameld tijdens het voltooien van de vereisten van deze tutorial.

    dbutils.widgets.text('source_file', "", "Source File")

spark.conf.set("fs.azure.account.auth.type", "OAuth")
spark.conf.set("fs.azure.account.oauth.provider.type", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set("fs.azure.account.oauth2.client.id", "<appId>")
spark.conf.set("fs.azure.account.oauth2.client.secret", "<password>")
spark.conf.set("fs.azure.account.oauth2.client.endpoint", "https://login.microsoftonline.com/<tenant>/oauth2/token")

adlsPath = 'abfss://data@contosoorders.dfs.core.windows.net/'
inputPath = adlsPath + dbutils.widgets.get('source_file')
customerTablePath = adlsPath + 'delta-tables/customers'

    Met deze code wordt een widget met de naam source_file gemaakt. Later gaat u een Azure-functie maken die deze code aanroept en een bestandspad aan die widget doorgeeft. Deze code authenticeert ook je serviceprincipal met het opslagaccount en creëert enkele variabelen die je in andere cellen gebruikt.

    Notitie

    In een productieomgeving kunt u de verificatiesleutel eventueel in Azure Databricks opslaan. Voeg vervolgens een zoeksleutel toe aan je codeblok in plaats van de authenticatiesleutel.

    Bijvoorbeeld, in plaats van deze regel code te gebruiken: spark.conf.set("fs.azure.account.oauth2.client.secret", "<password>"), gebruik de volgende regel code: spark.conf.set("fs.azure.account.oauth2.client.secret", dbutils.secrets.get(scope = "<scope-name>", key = "<key-name-for-service-credential>")).

    Nadat je deze tutorial hebt voltooid, zie je het artikel over Azure Data Lake Storage op de website van Azure Databricks om voorbeelden van deze aanpak te zien.

  2. Druk op SHIFT + ENTER om de code in dit blok uit te voeren.

  3. Kopieer en plak het volgende codeblok in een andere cel en druk vervolgens op SHIFT + ENTER om de code in dit blok uit te voeren.

    from pyspark.sql.types import StructType, StructField, DoubleType, IntegerType, StringType

inputSchema = StructType([
StructField("InvoiceNo", IntegerType(), True),
StructField("StockCode", StringType(), True),
StructField("Description", StringType(), True),
StructField("Quantity", IntegerType(), True),
StructField("InvoiceDate", StringType(), True),
StructField("UnitPrice", DoubleType(), True),
StructField("CustomerID", IntegerType(), True),
StructField("Country", StringType(), True)
])

rawDataDF = (spark.read
 .option("header", "true")
 .schema(inputSchema)
 .csv(adlsPath + 'input')
)

(rawDataDF.write
  .mode("overwrite")
  .format("delta")
  .saveAsTable("customer_data", path=customerTablePath))

    Deze code maakt de Databricks Delta-tabel aan in je opslagaccount en laadt vervolgens wat initiële gegevens uit het CSV-bestand dat je eerder hebt geüpload.

  4. Nadat dit codeblok succesvol is uitgevoerd, verwijder je dit codeblok uit je notebook.

Code toevoegen waarmee rijen worden ingevoegd in de Databricks Delta-tabel

  1. Kopieer en plak het volgende codeblok in een andere cel, maar voer deze cel nog niet uit.

    upsertDataDF = (spark
  .read
  .option("header", "true")
  .csv(inputPath)
)
upsertDataDF.createOrReplaceTempView("customer_data_to_upsert")

    Deze code voegt gegevens in een tijdelijke tabelweergave in door gebruik te maken van gegevens uit een CSV-bestand. Het pad naar dat CSV-bestand komt van de invoerwidget die je in een eerdere stap hebt aangemaakt.

  2. Kopieer en plak het volgende codeblok in een andere cel. Met deze code wordt de inhoud van de tijdelijke tabelweergave samengevoegd met de Databricks Delta-tabel.

    %sql
MERGE INTO customer_data cd
USING customer_data_to_upsert cu
ON cd.CustomerID = cu.CustomerID
WHEN MATCHED THEN
  UPDATE SET
    cd.StockCode = cu.StockCode,
    cd.Description = cu.Description,
    cd.InvoiceNo = cu.InvoiceNo,
    cd.Quantity = cu.Quantity,
    cd.InvoiceDate = cu.InvoiceDate,
    cd.UnitPrice = cu.UnitPrice,
    cd.Country = cu.Country
WHEN NOT MATCHED
  THEN INSERT (InvoiceNo, StockCode, Description, Quantity, InvoiceDate, UnitPrice, CustomerID, Country)
  VALUES (
    cu.InvoiceNo,
    cu.StockCode,
    cu.Description,
    cu.Quantity,
    cu.InvoiceDate,
    cu.UnitPrice,
    cu.CustomerID,
    cu.Country)

Een taak maken

Maak een taak aan die het eerder gemaakte notitieboek uitvoert. Later maak je een Azure-functie die deze taak uitvoert wanneer een event wordt opgeroepen.

  1. Selecteer Nieuwe>Job.

  2. Geef de baan een naam, kies het notitieboekje dat je hebt gemaakt en selecteer een cluster. Selecteer vervolgens Maken om de taak aan te maken.

    De nieuwe baan verschijnt in de lijst met banen samen met het notitieboekje en cluster dat je hebt gekozen.

Een Azure-functie maken

Maak een Azure-functie aan die de taak uitvoert.

  1. Selecteer in je Azure Databricks workspace je Azure Databricks-gebruikersnaam in de bovenste balk. Selecteer in de keuzelijst Gebruikersinstellingen.

  2. Selecteer op het tabblad Toegangstokens de optie Nieuw token genereren.

  3. Kopieer het token dat verschijnt en selecteer dan Klaar.

  4. Kies in de linkerbovenhoek van de Databricks-werkruimte het pictogram Personen en kies vervolgens Gebruikersinstellingen.

    Screenshot van het gebruikersinstellingenmenu voor het genereren van een Databricks-toegangstoken.

  5. Selecteer de knop Nieuw token genereren en selecteer vervolgens de knop Genereren .

    Zorg ervoor dat je het token kopieert naar een veilige plek. Je Azure-functie heeft dit token nodig om te authenticeren met Databricks zodat het de taak kan uitvoeren.

  6. Selecteer vanuit het menu van Azure Portal of op de startpagina de optie Een resource maken.

  7. Selecteer op de pagina NieuwCompute>Function App.

  8. Kies op de tabblad Basisbeginselen van de pagina Functie-app maken een resourcegroep en wijzig of controleer vervolgens de volgende instellingen:

    Instelling Waarde
    Function App-naam contosoorder
    Runtimestack .NET
    Publiceren Code
    Besturingssysteem Windows
    Abonnementtype Verbruik (serverloos)

  9. Selecteer Controleren + maken en selecteer vervolgens Maken.

    Wanneer de implementatie is voltooid, selecteert u Ga naar de resource om de overzichtspagina van de functie-app te openen.

  10. Selecteer Configuratie in de groep Instellingen.

  11. Kies op de pagina Toepassingsinstellingen de knop Nieuwe toepassingsinstelling om elke instelling toe te voegen.

    Screenshot van het toevoegen van een nieuwe applicatie-instelling in de Function App-configuratie.

    Voeg de volgende instellingen toe:

    Naam van de instelling Waarde
    DBX_INSTANCE De regio van uw Databricks-werkruimte. Bijvoorbeeld: westus2.azuredatabricks.net
    DBX_PAT Het persoonlijke toegangstoken dat u eerder hebt gegenereerd.
    DBX_JOB_ID De id van de taak die wordt uitgevoerd.

  12. Selecteer Opslaan om deze instellingen door te voeren.

  13. Selecteer in de groep Functions de optie Functies en selecteer vervolgens Maken.

  14. Kies Azure Event Grid-trigger.

    Installeer de extensie Microsoft.Azure.WebJobs.Extensions.EventGrid als u dit wordt gevraagd. Als je het moet installeren, selecteer dan opnieuw Azure Event Grid Trigger om de functie te maken.

    Het deelvenster Nieuwe functie wordt weergegeven.

  15. Voer in Nieuwe functie de functienaam in UpsertOrder en selecteer vervolgens Aanmaken.

  16. Vervang de inhoud van het codebestand door de volgende code en selecteer vervolgens Opslaan:

      #r "Azure.Messaging.EventGrid"
  #r "System.Memory.Data"
  #r "Newtonsoft.Json"
  #r "System.Text.Json"
  using Azure.Messaging.EventGrid;
  using Azure.Messaging.EventGrid.SystemEvents;
  using Newtonsoft.Json;
  using Newtonsoft.Json.Linq;

  private static HttpClient httpClient = new HttpClient();

  public static async Task Run(EventGridEvent eventGridEvent, ILogger log)
  {
     log.LogInformation("Event Subject: " + eventGridEvent.Subject);
     log.LogInformation("Event Topic: " + eventGridEvent.Topic);
     log.LogInformation("Event Type: " + eventGridEvent.EventType);
     log.LogInformation(eventGridEvent.Data.ToString());

     if (eventGridEvent.EventType == "Microsoft.Storage.BlobCreated" || eventGridEvent.EventType == "Microsoft.Storage.FileRenamed") {
        StorageBlobCreatedEventData fileData = eventGridEvent.Data.ToObjectFromJson<StorageBlobCreatedEventData>();
        if (fileData.Api == "FlushWithClose") {
              log.LogInformation("Triggering Databricks Job for file: " + fileData.Url);
              var fileUrl = new Uri(fileData.Url);
              var httpRequestMessage = new HttpRequestMessage {
                 Method = HttpMethod.Post,
                 RequestUri = new Uri(String.Format("https://{0}/api/2.0/jobs/run-now", System.Environment.GetEnvironmentVariable("DBX_INSTANCE", EnvironmentVariableTarget.Process))),
                 Headers = { 
                    { System.Net.HttpRequestHeader.Authorization.ToString(), "Bearer " + System.Environment.GetEnvironmentVariable("DBX_PAT", EnvironmentVariableTarget.Process)},
                    { System.Net.HttpRequestHeader.ContentType.ToString(), "application/json" }
                 },
                 Content = new StringContent(JsonConvert.SerializeObject(new {
                    job_id = System.Environment.GetEnvironmentVariable("DBX_JOB_ID", EnvironmentVariableTarget.Process),
                    notebook_params = new {
                          source_file = String.Join("", fileUrl.Segments.Skip(2))
                    }
                 }))
              };
              var response = await httpClient.SendAsync(httpRequestMessage);
              response.EnsureSuccessStatusCode();
        }
     }
  }

    Deze code parseert informatie over het opslag-event dat is opgewekt en maakt vervolgens een verzoekbericht met de URL van het bestand dat het event heeft getriggerd. Als onderdeel van het bericht geeft de functie een waarde door aan de widget source_file die u eerder hebt gemaakt. De functiecode stuurt het bericht naar de Databricks-taak en gebruikt het token dat je eerder hebt verkregen als authenticatie.

Een Event Grid-abonnement maken

In deze sectie maak je een Event Grid-abonnement aan dat de Azure-functie aanroept wanneer bestanden naar het opslagaccount worden geüpload.

  1. Selecteer Integratie. Selecteer op de pagina IntegratieEvent Grid Trigger.

  2. Geef in het deelvenster Trigger bewerken de gebeurtenis eventGridEvent een naam en selecteer vervolgens Gebeurtenisabonnement maken.

    Notitie

    De naam eventGridEvent komt overeen met de parameternaam die de Azure-functie ontvangt.

  3. Wijzig of controleer de volgende instellingen op het tabblad Basis van de pagina Gebeurtenisabonnement maken:

    Instelling Waarde
    Naam contoso-order-event-subscription
    Onderwerptype Opslagaccount
    Bron contosoorders
    Naam van systeemonderwerp <create any name>
    Filteren op gebeurtenistypen Blob gemaakt en blob verwijderd

  4. Klik op Creëren.

Het Event Grid-abonnement testen

  1. Maak een bestand met de naam customer-order.csv, plak de volgende gegevens in dat bestand en sla het op uw lokale computer op.

    InvoiceNo,StockCode,Description,Quantity,InvoiceDate,UnitPrice,CustomerID,Country
536371,99999,EverGlow Single,228,1/1/2018 9:01,33.85,20993,Sierra Leone

  2. Upload dit bestand in de Storage browser naar de invoermap van je storage-account.

    Wanneer je een bestand uploadt, activeert dit de gebeurtenis Microsoft.Storage.BlobCreated. Event Grid waarschuwt alle abonnees voor die gebeurtenis. In dit geval is de Azure-functie de enige abonnee. De Azure-functie parseert de gebeurtenisparameters om te bepalen welke gebeurtenis heeft plaatsgevonden. Vervolgens geeft het de URL van het bestand door aan de Databricks-taak. De Databricks-taak leest het bestand en voegt een rij toe aan de Databricks Delta-tabel die zich in je opslagaccount bevindt.

  3. Om te controleren of de taak is geslaagd, bekijk je de uitvoeringen van je taak. Je ziet een voltooiingsstatus. Voor meer informatie over hoe je runs voor een klus kunt bekijken, zie Bekijk runs voor een klus.

  4. Voer in een nieuwe werkboekcel deze query uit om de bijgewerkte delta-tabel te zien.

    %sql select * from customer_data

    De geretourneerde tabel bevat de meest recente record.

    Screenshot van de Databricks Delta-tabelquery die het laatste record toont.

  5. Als u deze record wilt bijwerken, maakt u een bestand met de naam customer-order-update.csv, plakt u de volgende gegevens in dat bestand en slaat u het op uw lokale computer op.

    InvoiceNo,StockCode,Description,Quantity,InvoiceDate,UnitPrice,CustomerID,Country
536371,99999,EverGlow Single,22,1/1/2018 9:01,33.85,20993,Sierra Leone

    Dit CSV-bestand is vrijwel identiek aan het vorige, behalve dat het aantal bestellingen is gewijzigd van 228 naar 22.

  6. Upload dit bestand in de Storage browser naar de invoermap van je storage-account.

  7. Voer de query select opnieuw uit om de bijgewerkte Delta-tabel weer te geven.

    %sql select * from customer_data

    De geretourneerde tabel toont de bijgewerkte record.

    Screenshot van de Databricks Delta-tabel query die het bijgewerkte record toont.

Hulpbronnen opschonen

Wanneer u de resources niet meer nodig hebt, verwijdert u de resourcegroep en alle gerelateerde resources. Om de resourcegroep te verwijderen, selecteer je de resourcegroep voor het opslagaccount en selecteer je Verwijderen.

Volgende stappen

Reageren op gebeurtenissen van Blob Storage

  • Last updated on