Copie várias tabelas incrementalmente com uma tarefa For each

Quando você precisa copiar dados de muitas tabelas de origem para tabelas do Unity Catalog em uma programação, copiar todas as linhas em cada execução é lento e caro. Use uma *marca d'água* para rastrear a última linha processada para cada tabela e copiar apenas novas linhas em cada execução.

Este tutorial mostra como criar um Job orientado por metadados que:

• Armazena a lista de tabelas de origem e o estado da marca d'água em uma tabela de controle Delta
• Usa uma tarefa For each para processar cada tabela em paralelo
• Copia apenas as linhas adicionadas desde a última execução bem-sucedida
• Atualiza a marca d'água após cada cópia bem-sucedida.

Como funciona

O Job usa três tipos de tarefa conectados em sequência:

Tarefa	Tipo	O que faz
read_watermarks	SQL	Lê a tabela de controle de marca d'água e retorna uma linha por tabela de origem
copy_tables	Para cada	Itera sobre {{tasks.read_watermarks.output.rows}}, executando a tarefa aninhada uma vez por tabela de origem
copy_incremental (aninhado)	Notebook	Lê linhas adicionadas desde a última marca d'água, as grava na tabela de destino e avança a marca d'água

Tarefa	Tipo	O que faz
read_watermarks	SQL	Lê a tabela de controle de marca d'água e retorna uma linha por tabela de origem
copy_tables	Para cada	Itera sobre {{tasks.read_watermarks.output.rows}}, executando a tarefa aninhada uma vez por tabela de origem
copy_incremental (aninhado)	Notebook	Lê linhas adicionadas desde a última marca d'água, as grava na tabela de destino e avança a marca d'água

A saída da tarefa SQL — um array JSON de objetos de linha — flui para o campo Entradas da tarefa For each usando {{tasks.read_watermarks.output.rows}}. O Notebook aninhado recebe source_table, target_table, watermark_column e last_watermark para cada iteração.

Pré-requisitos

• Um workspace Databricks com permissão para criar Job e Notebook
• Permissão para criar tabelas no Unity Catalog
• Um esquema do Unity Catalog onde você pode criar a tabela de controle e tabelas de destino (por exemplo, config)
• Um SQL warehouse para executar tarefas de SQL
• Tabelas de origem que contêm uma coluna monotonicamente crescente, como um carimbo de data/hora ou sequência de inteiros

Etapa 1: Criar a tabela de controle de marca d'água

A tabela de controle de marca d'água é a fonte da verdade para quais tabelas processar e até onde cada tabela foi copiada. Cada linha representa uma tabela de origem.

Execute o seguinte SQL para criar a tabela de controle e registrar duas tabelas de origem:

SQL

CREATE OR REPLACE TABLE config.watermarks (
  source_table      STRING NOT NULL,
  target_table      STRING NOT NULL,
  watermark_column  STRING NOT NULL,
  last_watermark    TIMESTAMP NOT NULL
);

INSERT INTO config.watermarks VALUES
  ('sales.raw_orders',    'sales.orders',    'updated_at', '1970-01-01'),
  ('sales.raw_customers', 'sales.customers', 'updated_at', '1970-01-01');

Configurar last_watermark para 1970-01-01 na primeira execução faz com que o Notebook copie todas as linhas existentes, agindo como uma carga inicial completa. As execuções subsequentes copiam apenas as linhas adicionadas ou atualizadas após a execução anterior.

O passo 2: Escreva o Notebook de cópia

O Notebook executa uma vez por iteração da tabela. Ele lê a marca d'água, filtra a origem, grava no destino e avança a marca d'água.

Crie um Notebook em um caminho como /Workspace/Users/<username>/copy_incremental e adicione o código a seguir:

Python

# Set defaults for running the notebook outside a job
dbutils.widgets.text("source_table",     "sales.raw_orders",  "Source table")
dbutils.widgets.text("target_table",     "sales.orders",      "Target table")
dbutils.widgets.text("watermark_column", "updated_at",        "Watermark column")
dbutils.widgets.text("last_watermark",   "1970-01-01",        "Last watermark")

source_table     = dbutils.widgets.get("source_table")
target_table     = dbutils.widgets.get("target_table")
watermark_column = dbutils.widgets.get("watermark_column")
last_watermark   = dbutils.widgets.get("last_watermark")

# Read only new rows from the source table
new_rows = spark.table(source_table).filter(
    f"{watermark_column} > '{last_watermark}'"
)

row_count = new_rows.count()
print(f"Copying {row_count} new rows from {source_table}")

if row_count > 0:
    # Append new rows to the target table, creating it if it does not exist
    new_rows.write.format("delta").mode("append").saveAsTable(target_table)

    # Compute the new high-water mark from the rows just written
    from pyspark.sql.functions import max as spark_max
    new_watermark = new_rows.agg(spark_max(watermark_column)).collect()[0][0]

    # Advance the watermark so the next run starts from here
    spark.sql(f"""
        UPDATE config.watermarks
        SET last_watermark = CAST('{new_watermark}' AS TIMESTAMP)
        WHERE source_table = '{source_table}'
    """)

    print(f"Watermark for {source_table} advanced to {new_watermark}")
else:
    print(f"No new rows for {source_table}, watermark unchanged")

Os padrões do dbutils.widgets.text() permitem executar e testar o Notebook diretamente. Quando o Notebook executa dentro da tarefa For each, o Job substitui estes defaults pelos valores reais para cada iteração.

nota

Este notebook usa o modo append, que é adequado quando a origem contém apenas inserções. Se sua origem contiver atualizações, use uma instrução MERGE em vez de write.mode("append") para fazer upsert de linhas na tabela de destino. Consulte Entre em uma tabela do Delta Lake usando merge para obter a sintaxe de merge.

Passo 3: Criar o Job

No seu workspace do Databricks, clique em fluxos de trabalho na barra lateral e depois clique em Criar Job . Atribua ao Job um nome como Incremental table copy.

Passo 4: Configure a tarefa de pesquisa de marca d'água

A tarefa SQL lê a tabela de controle e disponibiliza o resultado para a tarefa For each.

1. Clique em Adicionar tarefa .

2. Definir Nome da tarefa como read_watermarks.

3. Set Type to SQL .

4. No campo SQL , insira:

   SQL

   SELECT source_table, target_table, watermark_column, last_watermark
   FROM config.watermarks

5. Defina **SQL warehouse** para um warehouse em seu workspace.

6. Clique em Criar tarefa .

Quando esta tarefa é executada, o Databricks captura o resultado como uma matriz JSON em tasks.read_watermarks.output.rows:

JSON

[
  {
    "source_table": "sales.raw_orders",
    "target_table": "sales.orders",
    "watermark_column": "updated_at",
    "last_watermark": "2024-06-01T12:00:00.000Z"
  },
  {
    "source_table": "sales.raw_customers",
    "target_table": "sales.customers",
    "watermark_column": "updated_at",
    "last_watermark": "2024-06-01T12:00:00.000Z"
  }
]

Passo 5: Configurar a tarefa For each

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

1. Clique em Adicionar tarefa e defina Depende de como read_watermarks.

2. Definir Nome da tarefa como copy_tables.

3. Defina o **Tipo** como **Para cada**.

4. No campo **Entradas**, insira:

   
   {{tasks.read_watermarks.output.rows}}

5. Defina **Concorrência** como 2 para copiar duas tabelas por vez. Aumente este valor se seu warehouse puder suportar maior paralelismo.

6. Clique em **Adicionar uma tarefa para fazer loop** para configurar a tarefa aninhada.

7. Definir Nome da tarefa como copy_incremental.

8. Set Type to Notebook .

9. Defina **Caminho** como o caminho do Notebook que você criou no Passo 2.

10. Clique em Parâmetros , então clique em Adicionar para adicionar cada um dos seguintes parâmetros:

Chave	Valor
source_table	{{input.source_table}}
target_table	{{input.target_table}}
watermark_column	{{input.watermark_column}}
last_watermark	{{input.last_watermark}}

Chave	Valor
source_table	{{input.source_table}}
target_table	{{input.target_table}}
watermark_column	{{input.watermark_column}}
last_watermark	{{input.last_watermark}}

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

Passo 6: Execute o Job e verifique

1. Clique em Executar agora para acionar o Job.
2. Na página de execução do Job, clique no nó copy_tables para expandir a tarefa For each.
3. A página de execução mostra uma tabela de iterações — uma linha por tabela de origem — cada uma exibindo seu status, horário de início e duração.
4. Clique em qualquer iteração para exibir a saída do Notebook e confirmar a contagem de linhas e a atualização da marca d'água.

Para confirmar que a marca d'água avançou, execute a seguinte query após o Job ser concluído:

SQL

SELECT source_table, last_watermark FROM config.watermarks;

Cada valor last_watermark agora deve refletir o carimbo de data/hora da linha copiada mais recentemente. Se um valor ainda for 1970-01-01, a tabela de origem não continha linhas correspondentes ao filtro, ou a tarefa de cópia encontrou um erro — verifique a saída da execução da tarefa para detalhes.

Estender o padrão

Adicionar uma nova tabela de origem : insira uma linha na tabela de controle. A próxima execução do Job o coleta automaticamente, começando com uma carga completa de 1970-01-01:

SQL

INSERT INTO config.watermarks VALUES
  ('sales.raw_products', 'sales.products', 'updated_at', '1970-01-01');

**Pausar uma tabela**: Adicione uma active coluna e filtre na tarefa de SQL:

SQL

ALTER TABLE config.watermarks ADD COLUMN active BOOLEAN DEFAULT TRUE;

-- In the SQL task:
SELECT source_table, target_table, watermark_column, last_watermark
FROM config.watermarks
WHERE active = TRUE

Retroalimentar uma tabela : Reset sua marca d'água para copiar novamente de um ponto específico:

SQL

UPDATE config.watermarks
SET last_watermark = '2024-01-01'
WHERE source_table = 'sales.raw_orders';

Recursos adicionais

• Use uma tarefa For each para executar outra tarefa em um loop — Referência completa para configurar tarefas For each, incluindo opções de simultaneidade
• Use uma tabela de controle para acionar um For each Job — Acione um For each Job de uma tabela de configuração ativa
• Inserir ou atualizar em uma tabela Delta Lake usando merge — Use MERGE para inserir ou atualizar linhas quando os dados de origem incluírem atualizações
• As APIs AUTO CDC: Simplifique a captura de dados de alterações (CDC) com pipelines — Use a captura de dados de alterações (CDC) para fontes que rastreiam inserções, atualizações e exclusões