Tutorial: Implemente o padrão de captura de dados no data lake para atualizar uma tabela Delta do Databricks

Este tutorial mostra como manipular eventos em uma conta de armazenamento que tem um namespace hierárquico.

Você constrói uma pequena solução que permite preencher uma tabela Delta do Databricks carregando um arquivo de valores separados por vírgulas (CSV) que descreve uma ordem de venda. Você constrói essa solução conectando uma assinatura do Event Grid, uma Função do Azure e um Job no Azure Databricks.

Neste tutorial, você:

Criar uma assinatura do Event Grid que invoque uma Função do Azure.
Criar uma função do Azure que recebe uma notificação de um evento e, em seguida, executa o trabalho no Azure Databricks.
Criar um trabalho do Databricks que insere uma ordem de cliente em uma tabela do Databricks Delta que está localizada na conta de armazenamento.

Você constrói essa solução em ordem inversa, começando pelo espaço de trabalho do Azure Databricks.

Pré-requisitos

Criar uma conta de armazenamento que tenha um namespace hierárquico (Azure Data Lake Storage). Este tutorial usa uma conta de armazenamento nomeada contosoorders.

Confira Criar uma conta de armazenamento para uso com o Azure Data Lake Storage.
Verifique se a sua conta de usuário tem a função Colaborador de Dados do Storage Blob atribuída a ela.
Crie uma entidade de serviço, crie um segredo do cliente e, em seguida, conceda à entidade de serviço acesso à conta de armazenamento.

Confira Tutorial: Conectar-se ao Azure Data Lake Storage (etapas 1 a 3). Depois de completar essas etapas, certifique-se de colar o ID do inquilino, o ID do aplicativo e os valores do segredo do cliente em um arquivo de texto. Você precisa desses valores logo.
Se você não tiver uma assinatura do Azure, crie uma conta gratuita antes de começar.

Criar uma ordem de venda

Primeiro, crie um arquivo CSV que descreva um pedido de venda e depois faça o upload desse arquivo para a conta de armazenamento. Depois, você usa os dados desse arquivo para preencher a primeira linha da sua tabela Delta do Databricks.

Acesse sua nova conta de armazenamento no portal do Azure.
Selecione Navegador de armazenamento>Contêineres de blob>Adicionar contêiner e crie um contêiner chamado dados.
No contêiner dados, crie um diretório chamado entrada.

Em um editor de texto, cole o texto a seguir.

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

Salve esse arquivo no seu computador local e nomeie data.csv.
No navegador de armazenamento, faça o upload desse arquivo para a pasta de entrada .

Criar um trabalho no Azure Databricks

Nesta seção, você realiza as seguintes tarefas:

Criar um espaço de trabalho do Azure Databricks.
Crie um notebook.
Criar e preencher uma tabela Delta do Databricks.
Adicionar código que insere linhas na tabela do Databricks Delta.
Crie um trabalho.

Criar um workspace do Azure Databricks

Nesta seção, você cria um espaço de trabalho no Azure Databricks usando o portal do Azure.

Criar um espaço de trabalho do Azure Databricks. Dê um nome ao espaço de trabalho contoso-orders. Consulte Criar um espaço de trabalho do Azure Databricks.
Criar um cluster. Dê um nome ao cluster customer-order-cluster. Consulte Criar um cluster.
Crie um notebook. Nomeie o notebook configure-customer-table e escolha Python como a linguagem padrão do notebook. Consulte Criar um notebook.

Criar e preencher uma tabela do Databricks Delta

No notebook criado, copie e cole o bloco de código a seguir na primeira célula, mas não execute esse código ainda.

Substitua os valores dos placeholders appId, password e tenant neste bloco de código pelos valores que você coletou ao concluir os pré-requisitos deste 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'
```
Esse código cria um widget chamado source_file. Posteriormente, você criará uma função do Azure que chama esse código e passa um caminho de arquivo para esse widget. Esse código também autentica a entidade de serviço com a conta de armazenamento e cria algumas variáveis que você poderá usar em outras células.

Observação

Em uma configuração de produção, considere armazenar sua chave de autenticação no Azure Databricks. Depois, adicione uma chave de consulta ao seu bloco de código em vez da chave de autenticação.

Por exemplo, em vez de usar esta linha de código: spark.conf.set("fs.azure.account.oauth2.client.secret", "<password>"), use a seguinte linha de código: spark.conf.set("fs.azure.account.oauth2.client.secret", dbutils.secrets.get(scope = "<scope-name>", key = "<key-name-for-service-credential>")).

Depois de concluir este tutorial, veja o artigo sobre Azure Data Lake Storage no site do Azure Databricks para ver exemplos dessa abordagem.
Pressione SHIFT + ENTER para executar o código neste bloco.

Copie e cole o seguinte bloco de código em uma célula diferente e então pressione SHIFT + ENTER para executar o código desse bloco.

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

Esse código cria a tabela Delta do Databricks na sua conta de armazenamento e então carrega alguns dados iniciais do arquivo CSV que você enviou anteriormente.

Depois que esse bloco de código for executado com sucesso, remova-o do seu caderno.

Adicionar código que insere linhas na tabela do Databricks Delta

Copie e cole o bloco de código a seguir em uma célula diferente, mas não execute essa célula.
```
upsertDataDF = (spark
  .read
  .option("header", "true")
  .csv(inputPath)
)
upsertDataDF.createOrReplaceTempView("customer_data_to_upsert")
```
Esse código insere dados em uma visualização de tabela temporária usando dados de um arquivo CSV. O caminho para esse arquivo CSV vem do widget de entrada que você criou em uma etapa anterior.

Copie e cole o seguinte bloco de código em uma célula diferente. Esse código mescla o conteúdo da exibição de tabela temporária com a tabela do 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)

Criar um trabalho

Crie um trabalho que execute o caderno que você criou antes. Depois, você cria uma Função Azure que executa esse trabalho quando um evento é gerado.

Selecione Novo>Emprego.
Dê um nome ao trabalho, escolha o caderno que você criou e selecione um cluster. Em seguida, selecione Criar para criar o trabalho.

O novo emprego aparece na lista de Empregos junto com o caderno e o cluster que você selecionou.

Criar uma Função do Azure

Crie uma Função Azure que execute o trabalho.

No seu espaço de trabalho Azure Databricks, selecione seu nome de usuário Azure Databricks na barra superior. Na lista suspensa, selecione Configurações do Usuário.
Na guia Tokens de acessoselecione Gerar novo token.
Copie o token que aparece e então selecione Feito.
No canto superior do workspace do Databricks, escolha o ícone de pessoas e, em seguida, escolha Configurações do usuário.
Selecione o botão Gerar novo token e escolha o botão Gerar.

Certifique-se de copiar o token para um local seguro. Sua Função Azure precisa que esse token autentique com Databricks para que possa rodar o trabalho.
No menu do portal do Azure ou na Página inicial, selecione Criar um recurso.
Na página Novo, selecione Computação>Aplicativo de Funções.
Na guia Noções básicas da página Criar aplicativo de funções, escolha um grupo de recursos e altere ou verifique as seguintes configurações:

Configuração Valor

Nome do aplicativo de funções contosoorder

Pilha de runtime .NET

Publicar Code

Sistema operacional Windows

Tipo de plano Consumo (sem servidor)
Selecione Examinar + Criar e, em seguida, selecione Criar.

Quando a implantação for concluída, selecione Ir para o recurso para abrir a página de visão geral do aplicativo de funções.
No grupo Configurações, selecione Configuração.

Configuração	Valor
Nome do aplicativo de funções	contosoorder
Pilha de runtime	.NET
Publicar	Code
Sistema operacional	Windows
Tipo de plano	Consumo (sem servidor)

Na página Configurações do Aplicativo, escolha o botão Nova configuração de aplicativo para adicionar cada configuração.

Captura de tela da adição de uma nova configuração de aplicativo na configuração do Function App.

Adicione as seguintes configurações:

Nome da configuração	Valor
DBX_INSTANCE	A região do workspace do Databricks. Por exemplo: `westus2.azuredatabricks.net`
DBX_PAT	O token de acesso pessoal que você gerou anteriormente.
DBX_JOB_ID	O identificador da tarefa em execução.

Selecione Salvar para confirmar essas configurações.
No grupo Funções, selecione Functions e, em seguida, selecione Criar.
Escolha o gatilho do Grade de Eventos do Azure.

Instale a extensão Microsoft.Azure.WebJobs.Extensions.EventGrid se for solicitado que você faça isso. Se precisar instalá-la, selecione novamente o Grade de Eventos do Azure Trigger para criar a função.

O painel Nova Função é exibido.
Em Nova Função, insira UpsertOrder o nome da função e depois selecione Criar.

Substitua o conteúdo do arquivo de código pelo seguinte código e, em seguida, selecione Salvar:

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

Esse código analisa informações sobre o evento de armazenamento que foi gerado e, em seguida, cria uma mensagem de solicitação com a URL do arquivo que desencadeou o evento. Como parte da mensagem, a função passa um valor para o widget source_file que você criou anteriormente. O código da função envia a mensagem para o trabalho do Databricks e usa o token que você obteve anteriormente como autenticação.

Criar uma assinatura na Grade de Eventos

Nesta seção, você cria uma assinatura do Event Grid que chama a Função Azure quando arquivos são carregados na conta de armazenamento.

Selecione Integração. Na página Integração, selecione Gatilho de Grade de Eventos.
No painel Editar gatilho, nomeie o evento eventGridEvent e selecione Criar assinatura de evento.

Observação

O nome eventGridEvent corresponde ao nome do parâmetro que a Função Azure recebe.

Na guia Noções básicas da página Criar assinatura de evento, altere ou verifique as seguintes configurações:

Configuração	Valor
Nome	contoso-order-event-subscription
Tipo de tópico	Conta de armazenamento
Recurso de Origem	contosoorders
Nome do tópico do sistema	`<create any name>`
Filtro para Tipos de Evento	Blob criado e Blob excluído

Selecione Criar.

Testar a assinatura do Event Grid

Crie um arquivo chamado customer-order.csv, cole as informações a seguir nesse arquivo e salve-o no computador local.

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

No navegador de armazenamento, faça upload desse arquivo para a pasta de entrada da sua conta de armazenamento.

Quando você carrega um arquivo, ele dispara o evento Microsoft.Storage.BlobCreated. O Event Grid notifica todos os assinantes desse evento. Nesse caso, a Função Azure é a única assinante. A função do Azure analisa os parâmetros do evento para determinar qual evento ocorreu. Em seguida, passa a URL do arquivo para o job do Databricks. O trabalho do Databricks lê o arquivo e adiciona uma linha à tabela do Databricks Delta que está localizada na conta de armazenamento.
Para verificar se o trabalho foi concluído com sucesso, consulte as execuções do trabalho. Você vê um status de conclusão. Para mais informações sobre como visualizar execuções de um trabalho, consulte Visualizar execuções de um trabalho.
Em uma nova célula do livro de exercícios, execute esta consulta para ver a tabela delta atualizada.
```
%sql select * from customer_data
```
A tabela retornada mostra o registro mais recente.
Para atualizar esse registro, crie um arquivo chamado customer-order-update.csv, cole as informações a seguir nesse arquivo e salve-o no computador local.
```
InvoiceNo,StockCode,Description,Quantity,InvoiceDate,UnitPrice,CustomerID,Country
536371,99999,EverGlow Single,22,1/1/2018 9:01,33.85,20993,Sierra Leone
```
Este arquivo CSV é quase idêntico ao anterior, exceto que a quantidade da ordem é alterada de 228 para 22.
No navegador de armazenamento, faça upload desse arquivo para a pasta de entrada da sua conta de armazenamento.
Execute a consulta select novamente para ver a tabela delta atualizada.
```
%sql select * from customer_data
```
A tabela retornada mostra o registro atualizado.

Limpar os recursos

Quando você não precisar mais dos recursos, exclua o grupo de recursos e todos os recursos relacionados. Para excluir o grupo de recursos, selecione o grupo de recursos da conta de armazenamento e selecione Excluir.

Próximas etapas

Responder a eventos de armazenamento de blobs

Comentários

Esta página foi útil?

Last updated on 2026-07-25