Uso de una tabla de control para controlar un For each trabajo

Cuando ejecutas el mismo procesamiento en muchas entradas, como mercados, tablas fuente, clientes o particiones de fecha, codificar esa lista en tu trabajo implica editar código y volver a desplegar el trabajo cada vez que cambia la lista. En su lugar, almacena la lista en una tabla de control que el trabajo lee en tiempo de ejecución. Para añadir o eliminar trabajo, actualizas una fila en la tabla y la siguiente ejecución del trabajo recoge el cambio sin editar el trabajo en sí. Este es un patrón impulsado por metadatos : los datos, no el código, controlan lo que procesa el trabajo.

Este tutorial construye un trabajo que utiliza este patrón en el conjunto de datos de ejemplo preinstalado de Wanderbricks, para que puedas ejecutarlo de principio a fin sin crear ningún dato fuente. El escenario es una plataforma de alquiler vacacional que realiza el mismo análisis de precios para cada segmento de propiedad (como Ski Resort o Urban Year-Round). Una tabla de control lista los segmentos a analizar, una tarea SQL lee esa tabla y una For each tarea ejecuta el análisis una vez por segmento, en paralelo.

Cómo funciona el trabajo de la tabla de control

El trabajo conecta tres tareas en secuencia:

El flujo es read_segments → process_segments → run_segment_analysis (una vez por fila). La salida de la tarea SQL, un array JSON de objetos de fila, fluye al campo Entradas de la tarea For each a través de la referencia {{tasks.read_segments.output.rows}} de valor dinámico. La For each tarea pasa entonces los campos de cada fila a la tarea anidada como parámetros, disponibles como {{input.property_type}} y {{input.min_price}}.

Prerrequisitos

• Un área de trabajo de Azure Databricks con permiso para crear trabajos y cuadernos.
• Permiso para crear tablas en el Catálogo de Unity, y permiso para crear un esquema en un catálogo (los privilegios USE CATALOG y CREATE SCHEMA) para contener la tabla de control.
• Un almacén de SQL para ejecutar las tareas SQL. Si no tienes uno, consulta cómo crear un almacén SQL.
• El catálogo samples, que está disponible en todos los espacios de trabajo habilitados para Unity Catalog. El tutorial lee de samples.wanderbricks.properties, así que no hay datos fuente que configurar.

Paso 1: Crear la tabla de control

La tabla de control es la fuente de verdad para la lista de segmentos que procesa tu tarea. Para cambiar lo que hace la tarea, actualizas esta tabla, no la tarea.

Ejecuta el siguiente SQL en un cuaderno de Azure Databricks o en el editor SQL. La primera instrucción crea un esquema para contener la tabla de control, y la segunda crea la tabla con una fila por segmento de propiedad y el precio mínimo de venta que debe incluir en el análisis de ese segmento:

USE CATALOG <catalog-name>;

CREATE SCHEMA IF NOT EXISTS config;

CREATE OR REPLACE TABLE config.property_segments AS
SELECT * FROM VALUES
  ('Urban Year-Round', 150),
  ('Summer Getaway', 200),
  ('Ski Resort', 250)
AS t(property_type, min_price);

Sustituye <catalog-name> por un catálogo en el que puedas crear esquemas, como el catálogo de tu espacio de trabajo. Usa el mismo catálogo en todas las partes a las que hace referencia el tutorial config.property_segments, incluyendo la consulta de búsqueda en el Paso 3.

Después de este paso, config.property_segments contiene tres filas, una por segmento. Cada fila contiene los dos valores que el trabajo pasa a cada iteración: el property_type que se va a analizar y el límite inferior min_price por el que se va a filtrar.

Paso 2: Escribe la lógica de análisis

La tarea anidada dentro de la tarea For each se ejecuta una vez por cada fila de la tabla de control, recibiendo el property_type y el min_price de esa fila como parámetros. Puedes escribir esta lógica como una tarea de notebook o una tarea SQL. Elige según la lógica de tu negocio:

• Utiliza una tarea de notebook cuando la lógica de iteración por iteración necesite código procedimental, múltiples lenguajes o librerías (por ejemplo, un paso de ciencia de datos o aprendizaje automático).
• Usa una tarea SQL cuando la lógica sea una única consulta o transformación que puedas expresar de forma declarativa. Una tarea SQL necesita un almacén de SQL.

Ambas variantes a continuación producen el mismo resultado: para el segmento que se está procesando, el número de anuncios con un precio igual o superior a su precio mínimo y su precio medio.

Tarea de cuaderno

Cree un nuevo cuaderno en una ruta como /Workspace/Users/<username>/run_segment_analysis. Este cuaderno se ejecuta una vez por cada iteración de la For each tarea, recibiendo un segmento diferente cada vez.

Agregue el código siguiente al cuaderno:

# Set default values so you can run the notebook on its own while developing.
# When the notebook runs inside a For each task, the job overrides these defaults.
dbutils.widgets.text("property_type", "Ski Resort", "Property type")
dbutils.widgets.text("min_price", "250", "Minimum price")

# Read the parameters passed by the For each task.
property_type = dbutils.widgets.get("property_type")
min_price = dbutils.widgets.get("min_price")

result = spark.sql(
    """
    SELECT :property_type AS property_type,
           COUNT(*) AS property_count,
           ROUND(AVG(base_price), 2) AS avg_price
    FROM samples.wanderbricks.properties
    WHERE property_type = :property_type
      AND base_price >= :min_price
    """,
    args={"property_type": property_type, "min_price": min_price},
)
display(result)

Tarea SQL

Una tarea SQL ejecuta una consulta guardada, así que crea y guarda la consulta de análisis en el editor SQL ahora. Lo asocias a la tarea anidada cuando configuras la For each tarea en el Paso 4.

1. En su área de trabajo de Azure Databricks, haga clic en Nuevo>Consulta para abrir el editor de SQL.

2. Escriba la siguiente consulta. Las tareas SQL hacen referencia a parámetros con la sintaxis :param_name, por lo que la consulta lee su segmento y el piso de precio a partir de los parámetros :property_type y :min_price.

   SELECT :property_type AS property_type,
          COUNT(*) AS property_count,
          ROUND(AVG(base_price), 2) AS avg_price
   FROM samples.wanderbricks.properties
   WHERE property_type = :property_type
     AND base_price >= :min_price;
   

3. Haz clic en el título New Query <date> en el encabezado de la pestaña de tu archivo SQL y ponle el nombre run_segment_analysis. Luego haz clic en Guardar para moverlo a una carpeta donde quieres guardarlo.

La tarea For each pasa los valores de cada iteración a los parámetros denominados :property_type y :min_price en tiempo de ejecución. A diferencia de los widgets de cuaderno, los parámetros con nombre SQL no soportan valores por defecto: si un parámetro no se pasa, la consulta falla con un error de resolución de parámetros.

Paso 3: Crear la consulta de búsqueda

La tarea de consulta lee la tabla de control a través de una consulta guardada. Como en el Paso 2, crea y guarda la consulta en el editor SQL ahora, y luego addícala a la tarea de búsqueda en el Paso 4.

1. En su área de trabajo de Azure Databricks, haga clic en Nuevo>Consulta para abrir el editor de SQL.

2. Introduce lo siguiente, usando el mismo catálogo que elegiste en el Paso 1:

   SELECT property_type, min_price FROM <catalog-name>.config.property_segments;
   

   El nombre está totalmente calificado porque el almacén de SQL que ejecuta esta consulta podría tener de forma predeterminada un catálogo diferente al que creó la tabla.

3. Haz clic en el título New Query <date> en el encabezado de la pestaña de tu archivo SQL y ponle el nombre read_segments. Luego haz clic en Guardar para moverlo a una carpeta donde quieres guardarlo.

Paso 4: Crear y configurar el job

Con ambas consultas guardadas, crea el trabajo y añade sus dos tareas: la tarea de consulta SQL que lee la tabla de control y la For each tarea que ejecuta el análisis de cada fila.

Crear el trabajo

En su área de trabajo de Azure Databricks, en la barra lateral, haga clic en Nuevo>Trabajo. Ponle a la tarea un nombre descriptivo, como Segment Analysis.

Configurar la tarea de consulta SQL

Esta tarea lee la tabla de control y pone sus filas a disposición de la tarea For each ejecutando la consulta read_segments que guardaste en el Paso 3.

1. Haga clic en el mosaico Consulta SQL para configurar la primera tarea. Si la tesela consulta SQL no está disponible, haz clic en Añadir otro tipo de tarea y busca consulta SQL.
2. Establezca Nombre de tarea en read_segments.
3. Si es necesario, seleccione Consulta SQL en el menú desplegable Tipo.
4. En el campo de consulta SQL , selecciona la read_segments consulta que guardaste en el Paso 3.
5. Establezca SQL Warehouse para un almacén en su área de trabajo.
6. Haga clic en Create task (Crear tarea).

Cuando esta tarea se ejecuta, Azure Databricks captura el resultado como un array JSON en tasks.read_segments.output.rows. La salida de la tarea SQL siempre se devuelve como un array JSON, así que no necesitas ninguna configuración adicional. La forma general de la referencia es tasks.<task-name>.output.rows, donde <task-name> coincide con el nombre de la tarea que has establecido. La salida es similar a esta:

[
  { "property_type": "Urban Year-Round", "min_price": 150 },
  { "property_type": "Summer Getaway", "min_price": 200 },
  { "property_type": "Ski Resort", "min_price": 250 }
]

Configurar la For each tarea

La tarea For each lee la salida de SQL e inicia una ejecución de tarea anidada por tabla de origen.

1. Haga clic en Añadir tarea y seleccione Para cada uno.

2. Establezca Nombre de tarea en process_segments.

3. Verifica que Depends on está configurado como read_segments.

4. En el campo Entradas , introduce el array de filas capturado por la tarea SQL:

   {{tasks.read_segments.output.rows}}
   

5. Configura la concurrencia en 2 para ejecutar dos iteraciones en paralelo. Aumente este valor cuando la tarea anidada admita un paralelismo superior.

6. Para completar esta tarea, haz clic en Añadir una tarea sobre la que iterar y configura la tarea anidada que se ejecuta en cada iteración.

La For each tarea y su tarea anidada se crean juntas como una sola tarea. Configura la tarea anidada según el tipo que elegiste en el Paso 2:

Tarea de cuaderno

1. Establezca Nombre de tarea en run_segment_analysis.

2. Establezca Tipo en Notebook.

3. Establezca Ruta de acceso en el cuaderno que creó en el paso 2.

4. Haz clic en Parámetros y luego haz clic en Añadir para añadir cada parámetro:

   • Clave: property_type, Valor: {{input.property_type}}
   • Clave: min_price, Valor: {{input.min_price}}

   Cada {{input.<key>}} referencia se resuelve al campo correspondiente de la fila de la iteración actual.

5. Haga clic en Crear tarea para crear la tarea For each y su tarea anidada a la vez.

Tarea SQL

Esta tarea ejecuta la run_segment_analysis consulta que guardaste en el Paso 2.

1. Establezca Nombre de tarea en run_segment_analysis.

2. Ponle Tipo en SQL y luego pon la tarea SQL en Consulta.

3. En el campo de consulta SQL , selecciona la run_segment_analysis consulta que guardaste en el Paso 2.

4. Establezca SQL Warehouse para un almacén en su área de trabajo.

5. Haz clic en Parámetros y luego haz clic en Añadir para añadir cada parámetro:

   • Clave: property_type, Valor: {{input.property_type}}
   • Clave: min_price, Valor: {{input.min_price}}

   Cada {{input.<key>}} referencia se resuelve al campo correspondiente de la fila de la iteración actual.

6. Haga clic en Crear tarea para crear la tarea For each y su tarea anidada a la vez.

Tu grafo de tareas ahora muestra read_segments que fluye hacia process_segments, con la tarea anidada dentro del nodo For each.

Paso 5: Ejecuta la tarea y verifica

1. Haga clic en Ejecutar ahora para desencadenar el trabajo.
2. Selecciona la pestaña Ejecuciones para ver la ejecución. La primera ejecución de un trabajo tarda unos minutos en iniciar el proceso; cuando se completa, aparece en la lista.
3. Haga clic en el nodo process_segments para expandir la tarea For each.
4. La página de ejecución muestra una tabla de iteraciones, una fila por segmento, cada una con su estado, hora de inicio y duración.
5. Haz clic en cualquier fila de iteración para abrir su salida y confirmar que ha analizado el segmento esperado.

Puedes ver los resultados de cada iteración de forma independiente. Si falla una iteración específica, solo puedes reejecutar esa iteración desde la página de ejecución del trabajo sin volver a ejecutar todo el trabajo.

Extender el patrón

Para añadir un segmento al análisis, inserta una fila en la tabla de control:

INSERT INTO <catalog-name>.config.property_segments VALUES ('Historical Place', 100);

La siguiente ejecución de trabajo incluye el nuevo segmento, sin cambios en la configuración del trabajo ni ediciones en el notebook.

Este mismo patrón funciona en cualquier caso en el que quieras que los datos impulsen la iteración:

• Procesamiento por cliente: una fila por identificador de cliente. La tarea anidada aplica transformaciones específicas del cliente o entrega a destinos específicos del cliente.
• Ingesta de tablas: una fila por nombre de tabla fuente. La tarea anidada lee e ingiere cada tabla.
• Procesamiento de reposición: una fila por partición de fecha. La tarea anidada reprocesa datos históricos de esa partición.
• Ejecución basada en banderas de funcionalidad: una fila por cada funcionalidad o experimento habilitados. La tarea anidada activa la lógica correspondiente.

Para dejar de procesar una fila sin eliminarla, añade tu propia columna a la tabla de control (como un active indicador) y filtra por él en la tarea de búsqueda SQL. Esta es una columna ordinaria que usted define y rellena; la tarea For each no tiene un concepto incorporado de ella. Primero añada la columna y, a continuación, establezca las filas existentes en TRUE:

ALTER TABLE <catalog-name>.config.property_segments ADD COLUMN active BOOLEAN;
UPDATE <catalog-name>.config.property_segments SET active = TRUE;

Luego filtra en la read_segments consulta para que solo las filas activas impulsen la iteración:

SELECT property_type, min_price FROM <catalog-name>.config.property_segments WHERE active = TRUE;

Recursos adicionales

• Usar una For each tarea para ejecutar otra tarea en un bucle: Referencia completa para configurar For each tareas, incluidos los tipos de parámetros y las opciones de simultaneidad
• Usar una tabla de búsqueda para matrices de parámetros grandes en una For each tarea: Cómo controlar matrices de parámetros grandes que superen el límite de valores de tarea de 48 KB
• Acceder a los valores de los parámetros desde una tarea: todos los métodos para acceder a los valores de los parámetros en notebooks, scripts de Python y tareas SQL
• Conjunto de datos de Wanderbricks: El conjunto de datos de muestra utilizado en este tutorial