Tutorial: Implementación del patrón de captura de Data Lake para actualizar una tabla de Databricks Delta

En este tutorial se muestra cómo controlar eventos en una cuenta de almacenamiento que tiene un espacio de nombres jerárquico.

Construyes una pequeña solución que te permite llenar una tabla Delta de Databricks subiendo un archivo de valores separados por comas (CSV) que describe una orden de venta. Construyes esta solución conectando una suscripción a Event Grid, una función de Azure y un trabajo en Azure Databricks.

En este tutorial, usted hará lo siguiente:

Cree una suscripción de Event Grid que llame a una función de Azure.
Creará una función de Azure que recibe una notificación de un evento y, a continuación, ejecuta el trabajo en Azure Databricks.
Cree un trabajo en Databricks que inserte un pedido de un cliente en una tabla Delta de Databricks ubicada en la cuenta de almacenamiento.

Construyes esta solución en orden inverso, empezando por el espacio de trabajo de Azure Databricks.

Requisitos previos

Cree una cuenta de almacenamiento que tenga un espacio de nombres jerárquico (Azure Data Lake Storage). En este tutorial se usa una cuenta de almacenamiento denominada contosoorders.

Consulte Crear una cuenta de almacenamiento para usarla con Azure Data Lake Storage.
Asegúrese de que la cuenta de usuario tiene asignado el rol Colaborador de datos de Storage Blob.
Cree una entidad de servicio, cree un secreto de cliente y, a continuación, conceda a la entidad de servicio acceso a la cuenta de almacenamiento.

Consulte Tutorial: Conexión a Azure Data Lake Storage (pasos 1 a 3). Después de completar estos pasos, asegúrate de pegar el ID del inquilino, el ID de la aplicación y los valores secretos del cliente en un archivo de texto. Necesitas esos valores pronto.
Si no tiene una suscripción a Azure, cree una cuenta gratuita antes de empezar.

Creación de un pedido de ventas

Primero, crea un archivo CSV que describa una orden de venta y luego sube ese archivo a la cuenta de almacenamiento. Más adelante, usas los datos de este archivo para llenar la primera fila de tu tabla Delta de Databricks.

Vaya a la nueva cuenta de almacenamiento en el portal de Azure.
Selecciona Storage browser>Contenedores de blobs>Agregar contenedor y crea un nuevo contenedor llamado data.
En el contenedor data, cree un directorio denominado input.

En un editor de texto, pegue el texto siguiente.

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

Guarda este archivo en tu ordenador local y ponle data.csv.
En el navegador de almacenamiento, sube este archivo a la carpeta de entrada .

Creación de un trabajo en Azure Databricks

En esta sección, realizas estas tareas:

Crear un área de trabajo de Azure Databricks.
Cree un cuaderno.
Crear y rellenar una tabla de Databricks Delta.
Agregar código que inserta filas en la tabla de Databricks Delta.
Creación de un trabajo.

Creación de un área de trabajo de Azure Databricks

En esta sección, creas un espacio de trabajo de Azure Databricks utilizando el portal de Azure.

Crear un área de trabajo de Azure Databricks. Nombra el espacio de trabajo contoso-orders. Consulte Creación de un área de trabajo de Azure Databricks.
Crear un clúster. Asigne el nombre customer-order-cluster al clúster. Consulte Creación de un clúster.
Cree un cuaderno. Asigne al cuaderno el nombre configure-customer-table y elija Python como lenguaje predeterminado del mismo. Consulte Creación de un cuaderno.

Creación y relleno de una tabla de Databricks Delta

En el cuaderno que ha creado, copie y pegue el siguiente bloque de código en la primera celda, pero no ejecute el código aún.

Sustituye los valores de los marcadores de posición appId, password y tenant de este bloque de código por los valores que has recopilado al completar los requisitos previos de este 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'
```
Este código crea un widget denominado source_file. Más adelante, crearás una función de Azure que llama a este código y le pasa una ruta de archivo a ese componente. Este código también autentica tu entidad de servicio con la cuenta de almacenamiento y crea algunas variables que utilizarás en otras celdas.

Nota:

En una configuración de producción, considere la posibilidad de almacenar su clave de autenticación en Azure Databricks. Luego, añade una clave de consulta a tu bloque de códigos en lugar de la clave de autenticación.

Por ejemplo, en lugar de usar esta línea de código: spark.conf.set("fs.azure.account.oauth2.client.secret", "<password>"), utiliza la siguiente línea de código: spark.conf.set("fs.azure.account.oauth2.client.secret", dbutils.secrets.get(scope = "<scope-name>", key = "<key-name-for-service-credential>")).

Después de completar este tutorial, consulta el artículo de Azure Data Lake Storage en la web de Azure Databricks para ver ejemplos de este enfoque.
Pulsa SHIFT + ENTER para ejecutar el código de este bloque.

Copia y pega el siguiente bloque de código en una celda diferente, y luego pulsa SHIFT + ENTER para ejecutar el código de ese bloque.

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

Este código crea la tabla Delta de Databricks en tu cuenta de almacenamiento y luego carga algunos datos iniciales del archivo CSV que subiste antes.

Después de que este bloque de código funcione correctamente, elimínalo de tu cuaderno.

Agregar código que inserta filas en la tabla de Databricks Delta

Copie y pegue el siguiente bloque de código en otra celda, pero no ejecute la celda.
```
upsertDataDF = (spark
  .read
  .option("header", "true")
  .csv(inputPath)
)
upsertDataDF.createOrReplaceTempView("customer_data_to_upsert")
```
Este código inserta datos en una vista temporal de tabla utilizando datos de un archivo CSV. El camino hacia ese archivo CSV viene del widget de entrada que creaste en un paso anterior.

Copie y pegue el siguiente bloque de código en una celda distinta. Este código combina el contenido de la vista de tabla temporal con la tabla de 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)

Creación de un trabajo

Crea un trabajo que ejecute el cuaderno que creaste antes. Más adelante, creas una función de Azure que ejecuta este trabajo cuando se genera un evento.

Selecciona> trabajo.
Dale un nombre al trabajo, elige el cuaderno que has creado y selecciona un clúster. A continuación, seleccione Crear para crear el trabajo.

El nuevo trabajo aparece en la lista de Empleos junto con el cuaderno y el clúster que seleccionaste.

Creación de una Función de Azure

Crea una función de Azure que ejecute el trabajo.

En tu espacio de trabajo de Azure Databricks, selecciona tu nombre de usuario de Azure Databricks en la barra superior. Desde la lista desplegable, selecciona Configuración de usuario.
Seleccione Generar nuevo token en la pestaña Tokens de acceso.
Copia el token que aparece, y luego selecciona Hecho.
En la esquina superior del área de trabajo de Databricks, elija el icono de personas y, a continuación, seleccione Configuración de usuario.
Seleccione el botón Generar nuevo token y, después, el botón Generar.

Asegúrate de copiar el token en un lugar seguro. Tu función de Azure necesita que este token se autentique con Databricks para poder ejecutar el trabajo.
En el menú de Azure Portal o en la página Principal, seleccione Crear un recurso.
En la página Nuevo, seleccione Compute>Function App.

En la pestaña Aspectos básicos de la página Crear aplicación de funciones, elija un grupo de recursos y, a continuación, cambie o compruebe la siguiente configuración:

Configuración	Valor
Nombre de la aplicación de funciones	contosoorder
Pila de tiempo de ejecución	.NET
Publicar	Código
Sistema operativo	Windows
Tipo de plan	Consumo (sin servidor)

Seleccione Revisar y crear y, luego, Crear.

Cuando se complete la implementación, seleccione Ir al recurso para abrir la página Información general de la aplicación de funciones.
En el grupo Configuración, seleccione Configuración.

En la página Configuración de la aplicación, seleccione el botón Nueva configuración de la aplicación para agregar cada opción de configuración.

Captura de pantalla de añadir una nueva configuración de aplicación en la configuración de la App de Funcións.

Agregue la configuración siguiente:

Nombre de la configuración	Valor
DBX_INSTANCE	La región del espacio de trabajo de Databricks. Por ejemplo: `westus2.azuredatabricks.net`
DBX_PAT	El token de acceso personal que ha generado anteriormente.
DBX_JOB_ID	El identificador del trabajo en ejecución.

Seleccione Guardar para confirmar esta configuración.
En el grupo Funciones, seleccione Funciones y Crear.
Seleccione Azure Event Grid Trigger (Desencadenador de Azure Event Grid).

Instale la extensión Microsoft.Azure.WebJobs.Extensions.EventGrid si se le solicita. Si necesitas instalarla, selecciona de nuevo Azure Event Grid Trigger para crear la función.

Aparecerá el panel Nueva función.
En Nueva Función, escribe UpsertOrder el nombre de la función y luego selecciona Crear.

Sustituye el contenido del archivo de código por el siguiente código y luego seleccione Guardar:

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

Este código analiza información sobre el evento de almacenamiento que se generó y luego crea un mensaje de solicitud con la URL del archivo que activó el evento. Como parte del mensaje, la función pasa un valor al widget source_file que creó anteriormente. El código de función envía el mensaje al trabajo de Databricks y utiliza el token que obtuviste antes como autenticación.

Creación de una suscripción de Event Grid

En esta sección, creas una suscripción a Event Grid que llama a la Función de Azure cuando se suben archivos a la cuenta de almacenamiento.

Seleccione Integración. En la página Integración, selecciona Desencadenador de Event Grid.
En el panel Editar desencadenador, asigne al evento el nombre eventGridEvent y, a continuación, seleccione Crear suscripción de eventos.

Nota:

El nombre eventGridEvent coincide con el nombre del parámetro que recibe la función Azure.

En la pestaña Aspectos básicos de la página Crear suscripción de eventos, cambie o compruebe la siguiente configuración:

Configuración	Valor
Nombre	contoso-order-event-subscription
Tipo de tema	Cuenta de almacenamiento
Recurso de origen	contosoorders
Nombre del tema del sistema	`<create any name>`
Filtro para tipos de evento	Blob creado y Blob eliminado

Selecciona Crear.

Comprobación de la suscripción a Event Grid

Cree un archivo denominado customer-order.csv, pegue la siguiente información en ese archivo y guárdelo en el equipo local.

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

En el navegador de almacenamiento, sube este archivo a la carpeta de entrada de tu cuenta de almacenamiento.

Cuando cargas un archivo, se desencadena el evento Microsoft.Storage.BlobCreated. Event Grid notifica a todos los suscriptores de ese evento. En este caso, la función Azure es la única suscriptora. La función de Azure analiza los parámetros del evento para determinar qué evento se ha producido. Luego pasa la URL del archivo al trabajo de Databricks. El trabajo de Databricks lee el archivo y añade una fila a la tabla Delta de Databricks que está en tu cuenta de almacenamiento.
Para comprobar si el trabajo se ha ejecutado correctamente, consulta las ejecuciones de tu trabajo. Ves un estado de finalización. Para obtener más información sobre cómo ver las ejecuciones de un trabajo, consulte Ver ejecuciones de un trabajo.
En una nueva celda de libro de trabajo, ejecuta esta consulta para ver la tabla delta actualizada.
```
%sql select * from customer_data
```
La tabla devuelta muestra el registro más reciente.
Para actualizar este registro, cree un archivo denominado customer-order-update.csv, pegue la siguiente información en ese archivo y guárdelo en el equipo 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 archivo CSV es casi idéntico al anterior, salvo que la cantidad del pedido cambia de 228 a 22.
En el navegador de almacenamiento, sube este archivo a la carpeta de entrada de tu cuenta de almacenamiento.
Ejecute la consulta select de nuevo para ver la tabla delta actualizada.
```
%sql select * from customer_data
```
La tabla devuelta muestra el registro actualizado.

Limpieza de recursos

Cuando ya no necesite los recursos, elimine el grupo de recursos y todos los recursos relacionados. Para eliminar el grupo de recursos, selecciona el grupo de recursos para la cuenta de almacenamiento y selecciona Eliminar.

Pasos siguientes

Reacción a eventos de Blob Storage

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2026-07-25