Pular para o conteúdo principal
Última atualização em

Use uma tabela de controle para executar um trabalho For each

Quando você faz a execução do mesmo processamento em muitas entradas, como mercados, tabelas de origem, clientes ou partições de data, codificar essa lista diretamente no seu Job significa editar o código e reimplantar toda vez que a lista for alterada. Em vez disso, armazene a lista em uma tabela de controle que o job lê no momento da execução. Para adicionar ou remover trabalho, você atualiza uma linha na tabela, e a próxima execução do job detecta a alteração sem edições no próprio job. Este é um padrão orientado a metadados : os dados, e não o código, controlam o que o job processa.

Este tutorial cria um Job que usa esse padrão no dataset de amostra Wanderbricks pré-instalado, para que você possa fazer a execução dele de ponta a ponta sem criar nenhum dado de origem. O cenário é uma plataforma de aluguel de temporada que executa a mesma análise de preço para cada segmento de propriedade (como Ski Resort ou Urban Year-Round). Uma tabela de controle lista os segmentos a serem analisados, uma tarefa SQL lê essa tabela e uma tarefa For each executa a análise uma vez por segmento, em paralelo.

Como funciona​

O job conecta três tarefas em sequência:

Tarefa

Tipo

O que faz

read_segments

SQL

Lê a tabela de controle e captura as linhas como uma matriz JSON

process_segments

Para cada

Itera sobre a matriz de linhas, iniciando a tarefa aninhada uma vez por linha

run_segment_analysis

Notebook ou SQL (aninhado dentro de For each)

Executa uma vez por linha, usando os valores dessa linha para analisar um segmento de propriedade

Tarefa

Tipo

O que faz

read_segments

SQL

Lê a tabela de controle e captura as linhas como uma matriz JSON

process_segments

Para cada

Itera sobre a matriz de linhas, iniciando a tarefa aninhada uma vez por linha

run_segment_analysis

Notebook ou SQL (aninhado dentro de For each)

Executa uma vez por linha, usando os valores dessa linha para analisar um segmento de propriedade

O fluxo é read_segments → process_segments → run_segment_analysis (uma vez por linha). A saída da tarefa SQL, uma matriz JSON de objetos de linha, flui para o campo Inputs da tarefa For each por meio da referência de valor dinâmico {{tasks.read_segments.output.rows}}. A tarefa For each então passa os campos de cada linha para a tarefa aninhada como parâmetros, disponíveis como {{input.property_type}} e {{input.min_price}}.

Pré-requisitos​

  • Um workspace do Databricks com permissão para criar jobs e notebooks.
  • Permissão para criar tabelas no Unity Catalog e permissão para criar um esquema em um catálogo (os privilégios USE CATALOG e CREATE SCHEMA) para manter a tabela de controle.
  • Um SQL warehouse para a execução das tarefas de SQL. Se você não tiver um, consulte Criar um SQL warehouse.
  • O catálogo samples, que está disponível em todos os Workspace habilitados para o Unity Catalog. O tutorial faz a leitura a partir de samples.wanderbricks.properties, portanto, não há dados de origem para configurar.

O passo 1: Criar a tabela de controle​

A tabela de controle é a fonte da verdade para a lista de segmentos que seu job processa. Para alterar o que o job faz, você atualiza esta tabela, não o job.

Faça a execução do seguinte SQL em um notebook do Databricks ou no editor SQL. A primeira instrução cria um esquema para conter a tabela de controle, e a segunda cria a tabela com uma linha por segmento de propriedade e o preço mínimo de listagem a ser incluído na análise desse segmento:

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

Substitua <catalog-name> por um catálogo no qual você possa criar esquemas, como o catálogo do seu workspace. Use o mesmo catálogo em todos os locais em que o tutorial faz referência a config.property_segments, incluindo a query de pesquisa no o passo 3.

Após este o passo, config.property_segments contém três linhas, uma por segmento. Cada linha carrega os dois valores que o job passa para cada iteração: o property_type para analisar e o min_price floor para filtrar.

O passo 2: Escrever a lógica de análise​

A tarefa aninhada dentro da tarefa For each entra em execução uma vez por linha da tabela de controle, recebendo property_type e min_price dessa linha como parâmetros. Você pode escrever essa lógica como uma tarefa de notebook ou uma tarefa SQL. Escolha com base em sua lógica de negócios:

  • Use uma tarefa de notebook quando a lógica por iteração precisar de código procedural, vários idiomas ou bibliotecas (por exemplo, o passo de ciência de dados ou Machine Learning).
  • Use uma SQL tarefa quando a lógica for uma única query ou transformação que você possa expressar de forma declarativa. Uma tarefa SQL precisa de um SQL warehouse.

Ambas as variantes abaixo produzem o mesmo resultado: para o segmento sendo processado, o número de listagens no preço mínimo ou acima dele e seu preço médio.

Crie um novo notebook em um caminho como /Workspace/Users/<username>/run_segment_analysis. Este notebook é execução uma vez por iteração da tarefa For each, recebendo um segmento diferente a cada vez.

Adicione o seguinte código ao Notebook:

Python
# 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={&quot;property_type&quot;: property_type, &quot;min_price&quot;: min_price},
)
display(result)
nota

Chame dbutils.widgets.text() antes de dbutils.widgets.get(). Se você chamar get primeiro, a execução do notebook fora de um job gerará um erro InputWidgetNotDefined.

O passo 3: Criar a query de pesquisa​

A tarefa de pesquisa lê a tabela de controle por meio de uma query salva. Como no passo 2, crie e salve a query no editor SQL agora e, em seguida, anexe-a à tarefa de pesquisa no passo 4.

  1. No seu workspace do Databricks, clique em Ícone de mais. New > Query Icon. Query para abrir o editor SQL.

  2. Insira o seguinte, usando o mesmo catálogo que você escolheu no O passo 1:

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

    O nome está totalmente qualificado porque o SQL Warehouse que faz a execução desta query pode ter como default um catálogo diferente daquele em que você criou a tabela.

  3. Clique no título New Query <date> no cabeçalho da tab do seu arquivo SQL e dê a ele o nome read_segments. Em seguida, clique em Salvar para movê-lo para uma pasta onde você deseja armazená-lo.

O passo 4: Criar e configurar o Job​

Com ambas as queries salvas, crie o Job e adicione suas duas tarefas: a tarefa de pesquisa SQL que lê a tabela de controle e a tarefa For each de execução da análise para cada linha.

Criar o job​

No seu workspace do Databricks, na barra lateral, clique em Ícone de mais. Novo > Ícone de fluxos de trabalho. Job . Dê ao job um nome descritivo, como Segment Analysis.

Configurar a tarefa de pesquisa SQL​

Esta tarefa lê a tabela de controle e disponibiliza suas linhas para a tarefa For each executando a query read_segments que você salvou no Passo 3.

  1. Clique no bloco SQL query para configurar a primeira tarefa. Se o bloco SQL query não estiver disponível, clique em Add another tarefa type e pesquise por SQL query .
  2. Defina o nome da tarefa como read_segments.
  3. Se necessário, selecione SQL query no menu suspenso Tipo .
  4. No campo Query SQL , selecione a query read_segments que você salvou no passo 3.
  5. Configure SQL warehouse para um repositório em seu workspace.
  6. Clique em Criar tarefa .

Quando esta tarefa entra em execução, o Databricks captura o resultado como uma matriz JSON em tasks.read_segments.output.rows. A saída da tarefa SQL é sempre retornada como uma matriz JSON, portanto, você não precisa de nenhuma configuração extra. A forma geral da referência é tasks.<task-name>.output.rows, onde <task-name> corresponde ao nome da tarefa que você definiu. A saída tem a seguinte aparência:

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

Configure a tarefa For each​

A tarefa For each lê a saída SQL e inicia uma execução de tarefa aninhada por linha.

  1. Clique em Ícone de mais. Adicionar tarefa e selecione Para cada .

  2. Defina o nome da tarefa como process_segments.

  3. Verifique se Depends on está definido como read_segments.

  4. No campo Inputs , insira a matriz de linhas capturada pela tarefa SQL:


    {{tasks.read_segments.output.rows}}

  5. Defina Concorrência como 2 para execução de duas iterações em paralelo. Aumente este valor quando sua tarefa aninhada suportar maior paralelismo.

  6. Para concluir esta tarefa, clique em Adicionar uma tarefa para percorrer e configure a tarefa aninhada que é executada em cada iteração.

A tarefa For each e sua tarefa aninhada são criadas juntas como uma única tarefa. Configure a tarefa aninhada com base no tipo que você escolheu no o passo 2:

  1. Defina o nome da tarefa como run_segment_analysis.

  2. Set Type to Notebook .

  3. Defina o Caminho para o notebook que você criou na Etapa 2.

  4. Clique em Parâmetros e, em seguida, clique em Adicionar para adicionar cada parâmetro:

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

    Cada referência {{input.<key>}} é resolvida para o campo correspondente da linha da iteração atual.

  5. Clique em Criar tarefa para criar a tarefa For each e sua tarefa aninhada juntas.

O seu Job Directed Acyclic gráfico (DAG) agora mostra read_segments fluindo para process_segments, com a tarefa aninhada dentro do nó For each.

O passo 5: execução do Job e verificação​

  1. Clique em "Executar agora" para iniciar a tarefa.
  2. Selecione a tab Execuções para ver a execução. A primeira execução de um Job leva alguns minutos para começar o compute; quando ela é concluída, aparece na lista.
  3. Clique no nó process_segments para expandir a tarefa For each.
  4. A página de execução mostra uma tabela de iterações, uma linha por segmento, cada uma com seu status, horário de início e duração.
  5. Clique em qualquer linha de iteração para abrir sua saída e confirmar se ela analisou o segmento esperado.

Você pode ver os resultados de cada iteração de forma independente. Se uma iteração específica falhar, você poderá executar novamente apenas essa iteração a partir da página de execução do job, sem precisar executar o job inteiro novamente.

Amplie o padrão​

Para adicionar um segmento à análise, insira uma linha na tabela de controle:

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

A próxima execução do job inclui o novo segmento, sem alterações na configuração do job ou edições no notebook.

Este mesmo padrão funciona para qualquer caso em que você deseje que os dados impulsionem a iteração:

  • Processamento por cliente : uma linha por ID de cliente. A tarefa aninhada aplica transformações específicas do cliente ou entrega em destinos específicos do cliente.
  • Ingestão de tabela : uma linha por nome de tabela de origem. A tarefa aninhada lê e ingere cada tabela.
  • Processamento de backfill : uma linha por partição de data. A tarefa aninhada reprocessa a data histórica para essa partição.
  • Execução orientada por sinalizador de recurso : uma linha por recurso ou experimento habilitado. A tarefa aninhada ativa a lógica correspondente.

Para interromper o processamento de uma linha sem excluí-la, adicione sua própria coluna à tabela de controle (como uma flag active) e filtre-a na tarefa de pesquisa SQL. Esta é uma coluna comum que você define e preenche; a tarefa For each não possui um conceito integrado para ela. Primeiro adicione a coluna e, em seguida, defina as linhas existentes como TRUE:

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

Em seguida, filtre-o na query read_segments para que apenas as linhas ativas conduzam a iteração:

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

Recursos adicionais​

Nesta página