Självstudie: Implementera data lake capture-mönstret för att uppdatera en Databricks Delta-tabell

Den här handledningen visar hur du hanterar och arbetar med händelser i ett lagringskonto som har ett hierarkiskt namnområde.

Du bygger en liten lösning som låter dig fylla i en Databricks Delta-tabell genom att ladda upp en CSV-fil (komma-separerade värden) som beskriver en försäljningsorder. Du bygger denna lösning genom att koppla ihop en Event Grid-prenumeration, en Azure-funktion och ett jobb i Azure Databricks.

I den här handledningen kommer du att:

  • Skapa en Event Grid-prenumeration som anropar en Azure-funktion.
  • Skapa en Azure-funktion som tar emot ett meddelande från en händelse och kör sedan jobbet i Azure Databricks.
  • Skapa ett Databricks-jobb som infogar en kundorder i en Databricks Delta-tabell som finns i lagringskontot.

Du bygger denna lösning i omvänd ordning, med start i Azure Databricks arbetsyta.

Prerequisites

Skapa en försäljningsorder

Skapa först en CSV-fil som beskriver en försäljningsorder, och ladda sedan upp den filen till lagringskontot. Senare använder du datan från denna fil för att fylla i första raden i din Databricks Delta-tabell.

  1. Gå till ditt nya lagringskonto i Azure-portalen.

  2. Välj Storage browser>Blob containers>Lägg till container och skapa en ny container med namnet data.

    Skärmdump av att skapa en container i webbläsaren Azure Storage.

  3. Skapa en katalog med namnet input i datacontainern.

  4. Klistra in följande text i en textredigerare.

    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. Spara den här filen på din lokala dator och döp den tilldata.csv.

  6. I Storage Browser, ladda upp denna fil till inmatningsmappen .

Skapa ett jobb i Azure Databricks

I detta avsnitt utför du dessa uppgifter:

  • Skapa en Azure Databricks-arbetsyta.
  • Skapa en anteckningsbok.
  • Skapa och fylla i en Databricks Delta-tabell.
  • Lägg till kod som infogar rader i tabellen Databricks Delta.
  • Skapa ett jobb.

Skapa en Azure Databricks-arbetsyta

I detta avsnitt skapar du en Azure Databricks-arbetsyta genom att använda Azure-portalen.

  1. Skapa en Azure Databricks-arbetsyta. Namnge arbetsytan contoso-orders. Se Skapa en Azure Databricks-arbetsyta.

  2. Skapa ett kluster. Ge klustret namnet customer-order-cluster. Se Skapa ett kluster.

  3. Skapa en anteckningsbok. Namnge anteckningsboken configure-customer-table och välj Python som standardspråk för notebook-filen. Se Skapa en notebook.

Skapa och fylla i en Databricks Delta-tabell

  1. I anteckningsboken som du skapade kopierar och klistrar du in följande kodblock i den första cellen, men kör inte den här koden ännu.

    Ersätt platshållarvärdena appId, password och tenant i det här kodblocket med de värden som du samlade in när du slutförde förberedelserna för den här självstudiekursen.

    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'

    Den här koden skapar en widget med namnet source_file. Senare skapar du en Azure-funktion som anropar den här koden och skickar en filsökväg till widgeten. Denna kod autentiserar också din tjänsteprincipal med lagringskontot och skapar vissa variabler som du använder i andra celler.

    Note

    I en produktionsinställning bör du överväga att lagra din autentiseringsnyckel i Azure Databricks. Lägg sedan till en uppslagsnyckel i ditt kodblock istället för autentiseringsnyckeln.

    Till exempel, istället för att använda denna kodrad: spark.conf.set("fs.azure.account.oauth2.client.secret", "<password>"), använd följande kodrad: spark.conf.set("fs.azure.account.oauth2.client.secret", dbutils.secrets.get(scope = "<scope-name>", key = "<key-name-for-service-credential>")).

    Efter att du har slutfört denna handledning, se artikeln om Azure Data Lake Storage på Azure Databricks webbplats för exempel på detta tillvägagångssätt.

  2. Tryck på SHIFT + ENTER för att köra koden i detta block.

  3. Kopiera och klistra in följande kodblock i en annan cell, och tryck sedan på SHIFT + ENTER för att köra koden i detta block.

    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))

    Denna kod skapar Databricks Delta-tabellen i ditt lagringskonto och laddar sedan in initial data från CSV-filen som du laddade upp tidigare.

  4. När detta kodblock körs framgångsrikt, ta bort det från din anteckningsbok.

Lägg till kod som infogar rader i tabellen Databricks Delta

  1. Kopiera och klistra in följande kodblock i en annan cell, men kör inte den här cellen.

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

    Denna kod infogar data i en tillfällig tabellvy genom att använda data från en CSV-fil. Vägen till den CSV-filen kommer från inmatningswidgeten som du skapade i ett tidigare steg.

  2. Kopiera och klistra in följande kodblock i en annan cell. Den här koden sammanfogar innehållet i den tillfälliga tabellvyn med tabellen Databricks Delta.

    %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)

Skapa ett jobb

Skapa ett jobb som kör anteckningsboken du skapade tidigare. Senare skapar du en Azure-funktion som kör detta jobb när en händelse tas upp.

  1. Välj Nytt>jobb.

  2. Ge jobbet ett namn, välj den anteckningsbok du skapade och välj ett kluster. Välj sedan Skapa för att skapa jobbet.

    Det nya jobbet visas i Jobblistan tillsammans med anteckningsboken och klustret du valde.

Skapa en Azure-funktion

Skapa en Azure-funktion som kör jobbet.

  1. I din Azure Databricks workspace, välj ditt Azure Databricks-användarnamn i den övre fältet. Från rullgardinsmenyn väljer du Användarinställningar.

  2. På fliken Åtkomsttoken väljer du Generera ny token.

  3. Kopiera tokenen som visas och välj sedan Klart.

  4. I det övre hörnet av Databricks-arbetsytan väljer du personikonen och väljer sedan Användarinställningar.

    Skärmdump av användarinställningsmenyn för att generera en Databricks-åtkomsttoken.

  5. Välj knappen Generera ny token och välj sedan knappen Generera .

    Se till att kopiera tokenen till en säker plats. Din Azure-funktion behöver denna token för att autentisera sig med Databricks så att den kan köra jobbet.

  6. I menyn i Azure-portalen eller på sidan Start väljer du Skapa en resurs.

  7. På sidan Nytt väljer du Beräkna>Funktion App.

  8. På fliken Grundinställningar på sidan Skapa funktionsapp väljer du en resursgrupp och ändrar eller verifierar sedan följande inställningar:

    Inställning Värde
    Funktionsappens namn contosoorder
    Körningstack .NET
    Publicera Kod
    Operativsystem Windows
    Plantyp Förbrukning (serverlös)

  9. Välj Granska + skapaoch välj sedan Skapa.

    När distributionen är klar väljer du Gå till resurs för att öppna översiktssidan för funktionsappen.

  10. I gruppen Inställningar väljer du Konfiguration.

  11. På sidan Programinställningar väljer du knappen Ny programinställning för att lägga till varje inställning.

    Skärmdump av att lägga till en ny applikationsinställning i Funktionsapp-konfigurationen.

    Lägg till följande inställningar:

    Inställningsnamn Värde
    DBX_INSTANCE Regionen för din Databricks-arbetsyta. Till exempel: westus2.azuredatabricks.net
    DBX_PAT Den personliga åtkomsttoken som du genererade tidigare.
    DBX_JOB_ID Identifieraren för det jobb som körs.

  12. Välj Spara för att checka in de här inställningarna.

  13. I gruppen Funktioner väljer du Funktioner och sedan Skapa.

  14. Välj Azure Event Grid-utlösare.

    Installera tillägget Microsoft.Azure.WebJobs.Extensions.EventGrid om du uppmanas att göra det. Om du behöver installera det, välj Azure Event Grid Trigger igen för att skapa funktionen.

    Fönstret Ny funktion visas.

  15. I Ny funktion, ange UpsertOrder funktionsnamnet och välj sedan Skapa.

  16. Byt ut innehållet i kodfilen mot följande kod och välj sedan Spara:

      #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();
        }
     }
  }

    Denna kod tolkar information om lagringshändelsen som skapades och skapar sedan ett förfrågningsmeddelande med URL:en till filen som utlöste händelsen. Som en del av meddelandet skickar funktionen ett värde till den source_file widget som du skapade tidigare. Funktionskoden skickar meddelandet till Databricks-jobbet och använder token du fick tidigare som autentisering.

Skapa en Event Grid-prenumeration

I denna sektion skapar du en Event Grid-prenumeration som anropar Azure-funktionen när filer laddas upp till lagringskontot.

  1. Välj Integrering. På integrationssidan , välj Event Grid Trigger.

  2. I fönstret Redigera utlösare namnger du händelsen eventGridEventoch väljer sedan Skapa händelseprenumeration.

    Note

    Namnet eventGridEvent stämmer överens med parameternamnet som Azure-funktionen får.

  3. På fliken Grundläggande inställningar på sidan Skapa händelseprenumeration ändrar eller verifierar du följande inställningar:

    Inställning Värde
    Name contoso-order-event-subscription
    Ämnestyp Lagringskonto
    Källresurs contosoorders
    Systemämnesnamn <create any name>
    Filtrera efter händelsetyper Blob skapad och blob raderad

  4. Välj Skapa.

Testa Event Grid-prenumerationen

  1. Skapa en fil med namnet customer-order.csv, klistra in följande information i filen och spara den på den lokala datorn.

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

  2. I Storage Browser, ladda upp denna fil till inmatningsmappen på ditt lagringskonto.

    När du laddar upp en fil höjer den Microsoft. Storage.BlobCreated-händelsen. Event Grid meddelar alla prenumeranter på händelsen. I detta fall är Azure-funktionen den enda prenumeranten. Azure-funktionen parsar händelseparametrarna för att avgöra vilken händelse som inträffade. Den skickar sedan filens URL till Databricks-jobbet. Databricks-jobbet läser filen och lägger till en rad i Databricks Delta-tabellen som finns i ditt lagringskonto.

  3. För att kontrollera om jobbet lyckades, se körningarna för ditt jobb. Du ser status för slutförande. Mer information om hur du visar körningar för ett jobb finns i Visa körningar för ett jobb.

  4. I en ny arbetsbokscell, kör denna fråga för att se den uppdaterade delta-tabellen.

    %sql select * from customer_data

    Den returnerade tabellen visar den senaste posten.

    Skärmdump av Databricks Delta-tabellfråga som visar den senaste posten.

  5. Om du vill uppdatera den här posten skapar du en fil med namnet customer-order-update.csv, klistrar in följande information i filen och sparar den på den lokala datorn.

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

    Denna CSV-fil är nästan identisk med den föregående, förutom att beställningsvolymen ändras från 228 till .22

  6. I Storage Browser, ladda upp denna fil till inmatningsmappen på ditt lagringskonto.

  7. Kör frågan select igen för att se den uppdaterade deltatabellen.

    %sql select * from customer_data

    Tabellen som returnerades visar den uppdaterade posten.

    Skärmdump av Databricks Delta-tabellfråga som visar den uppdaterade posten.

Rensa resurser

När du inte längre behöver resurserna tar du bort resursgruppen och alla relaterade resurser. För att ta bort resursgruppen, välj resursgruppen för lagringskontot och välj Ta bort.

Nästa steg

Reagera på Blob Storage-händelser

  • Last updated on