As APIs AUTO CDC: captura simplificada de dados de mudanças com pipelines

Os pipelines do Lakeflow simplificam a captura de dados alterados (CDC) com as APIs AUTO CDC e AUTO CDC FROM SNAPSHOT. Estas APIs automatizam a complexidade de computar dimensões que mudam lentamente (SCD) Tipo 1 e Tipo 2, seja a partir de um feed CDC ou de instantâneos de bases de dados. A AUTO CDC API também suporta rastreamento bitemporal, que regista alterações ao longo de duas dimensões temporais (Beta). Para saber mais sobre SCD de Tipo 1 e de Tipo 2, consulte Captura de dados alterados e capturas instantâneas. Para mais informações sobre rastreamento bitemporal, veja Bitemporal AUTO CDC.

A API que usa depende da origem dos seus dados de alteração:

• AUTO CDC: Use isto quando a base de dados de origem tiver um feed CDC ativado. AUTO CDC processa alterações a partir de um feed de dados de alteração (CDF). É suportado tanto nas interfaces de SQL e Python do sistema de pipeline.
• AUTO CDC FROM SNAPSHOT: Use isto quando o CDC não estiver ativado na base de dados de origem e apenas os snapshots estiverem disponíveis. Esta API compara instantâneos para determinar alterações e depois processa-as. É suportado tanto nas interfaces de SQL e Python do sistema de pipeline.

Para alvos SCD Tipo 1, também pode combinar um fluxo AUTO CDC FROM SNAPSHOT com um ou mais fluxos AUTO CDC. Utilize este padrão quando uma fonte fornece instantâneos autoritativos periódicos e um feed CDC de menor latência entre instantâneos.

Ambas as APIs suportam a atualização de tabelas usando SCD Tipo 1 e Tipo 2:

• Use o SCD Tipo 1 para atualizar registos diretamente. O histórico não é retido para registros atualizados.
• Use o SCD Tipo 2 para manter um histórico de registos, seja em todas as atualizações ou em atualizações de um conjunto especificado de colunas.

Apenas para AUTO CDC, também é possível utilizar armazenamento bitemporal, que estende o histórico do SCD Tipo 2 para rastrear alterações em duas dimensões temporais: tempo de negócio e tempo de sistema. Bitemporal está em fase Beta. Consulte Bitemporal AUTO CDC.

AUTO CDC também suporta atualizações parciais, onde um registo de alteração atualiza apenas um subconjunto de colunas. Consulte Aplicar atualizações parciais.

As AUTO CDC APIs não são suportadas pelo Apache Spark Declarative Pipelines.

Para sintaxe e outras referências, veja AUTO CDC INTO (pipelines),create_auto_cdc_flow e create_auto_cdc_from_snapshot_flow.

Requerimentos

Para usar as APIs de CDC, o seu pipeline tem de estar configurado para usar pipelines Lakeflow sem servidor ou os pipelines do Lakeflow Pro ou Advancededições.

Como funciona o AUTO CDC

Para realizar o processamento CDC com AUTO CDC, crie uma tabela de streaming e depois use a instrução AUTO CDC ... INTO em SQL ou a função create_auto_cdc_flow() em Python para especificar a fonte, as chaves e o sequenciamento para o feed de alterações. Para uma explicação de como funcionam a sequenciação e a lógica SCD, veja Alterar captura de dados e instantâneos. Veja os exemplos do AUTO CDC.

Para a inicialização inicial de uma fonte com um feed de alterações, use AUTO CDC com um once fluxo e depois continue a processar o feed de alterações. Veja Replicar uma tabela RDBMS externa usando AUTO CDC.

Para detalhes de sintaxe, consulte AUTO CDC INTO (pipelines) ou create_auto_cdc_flow.

Como funciona o AUTO CDC do SNAPSHOT

AUTO CDC FROM SNAPSHOT determina alterações nos dados de origem comparando instantâneos por ordem. Você pode ler capturas instantâneas de uma tabela Delta, arquivos de armazenamento na cloud ou JDBC diretamente. É suportado tanto nas interfaces de SQL e Python do sistema de pipeline.

Para realizar o processamento CDC com AUTO CDC FROM SNAPSHOT, crie uma tabela de streaming e depois defina o fluxo:

• Em SQL, utilize a AUTO CDC ... FROM SNAPSHOT forma da instrução CREATE FLOW ou a cláusula embutida FLOW AUTO CDC de CREATE STREAMING TABLE. A origem é uma cláusula obrigatória FROM SNAPSHOT (snapshot_query) e uma cláusula opcional WITH VERSION (version_query) que seleciona a próxima versão snapshot a processar. Ver CREATE FLOW (pipelines).
• Em Python, use a função create_auto_cdc_from_snapshot_flow() para especificar o instantâneo, as chaves e outros argumentos. Veja create_auto_cdc_from_snapshot_flow.

Para obter detalhes sobre os dois padrões de ingestão e quando usar cada um, consulte Padrões de processamento de instantâneos. Veja os exemplos do AUTO CDC FROM SNAPSHOT.

Combine AUTO CDC FROM SNAPSHOT e AUTO CDC

A versão snapshot e a coluna de sequenciação CDC formam um domínio de ordenação. Um evento CDC mais recente do que o snapshot mais recente tem prioridade sobre o snapshot. Um snapshot mais recente tem prioridade sobre eventos CDC mais antigos e trata as chaves ausentes do snapshot como eliminadas na versão do snapshot. Eventos mais recentes do CDC podem preservar ou restaurar essas chaves.

Um alvo unificado deve cumprir os seguintes requisitos:

• O destino utiliza SCD Tipo 1.
• O destino tem exatamente um AUTO CDC FROM SNAPSHOT fluxo e um ou mais AUTO CDC fluxos.
• Cada fluxo AUTO CDC tem um nome único.
• Todos os fluxos usam o mesmo número de chaves na mesma ordem. Os nomes de chaves snapshot-flow são comparados sem distinção entre maiúsculas e minúsculas com os nomes de chaves AUTO CDC. Fluxos múltiplos AUTO CDC devem usar nomes-chave e capitalização idênticos.
• A versão snapshot e cada coluna de sequenciação CDC têm exatamente o mesmo tipo de dados.
• O AUTO CDC FROM SNAPSHOT fluxo não define expectativas.
• Os AUTO CDC fluxos não utilizam IGNORE NULL UPDATES nem as suas variantes de lista de colunas.
• O pipeline utiliza o modo acionado por evento.
• Se adicionar um AUTO CDC fluxo a um destino existente AUTO CDC FROM SNAPSHOT , o destino não deve conter colunas de utilizador cujos nomes entrem em conflito com as colunas reservadas AUTO CDC do sistema.

Tanto os flows AUTO CDC como os AUTO CDC FROM SNAPSHOT podem usar a interface de pipeline em Python ou SQL. Podes misturar fluxos SQL e Python no mesmo destino.

Utilize once=True em Python ou ONCE em SQL no AUTO CDC FROM SNAPSHOT fluxo para preencher um snapshot uma vez enquanto os AUTO CDC fluxos continuam a processar novos eventos em atualizações subsequentes. Uma atualização completa do destino executa novamente o fluxo de instantâneo único. Para um exemplo completo, veja Adicionar um backfill a uma tabela AUTO CDC SCD Tipo 1.

Este padrão não suporta destinos SCD Tipo 2 ou bitemporais.

Usar várias colunas para sequenciamento

Para sequenciar por múltiplas colunas (por exemplo, um carimbo temporal e um ID para desempatar), use a STRUCT para as combinar. A API ordena inicialmente pelo primeiro campo e, em caso de empate, considera o segundo campo, e assim sucessivamente.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Python

sequence_by = struct("timestamp_col", "id_col")

Exemplos AUTO CDC

Os exemplos seguintes demonstram o processamento SCD Tipo 1 e Tipo 2 usando uma fonte de dados de alteração. Os dados de exemplo criam novos registos de utilizador, apagam um registo de utilizador e atualizam os registos de utilizador. No exemplo SCD Tipo 1, as últimas UPDATE operações chegam tarde e são removidas da tabela de destino, demonstrando o manejo de eventos fora de ordem.

Seguem-se os registos de entrada usados nestes exemplos. Estes dados são criados ao executar a consulta na secção Criar dados de exemplo .

Se descomentar a última linha na consulta de geração de dados de exemplo, insere o seguinte registo que especifica truncar a tabela (limpar a tabela) em sequenceNum=3:

Criar dados de exemplo

Execute as seguintes instruções para criar um conjunto de dados de exemplo. Este código não se destina a ser executado como parte de uma definição de pipeline. Executa-o a partir da pasta de exploração do teu pipeline, em vez da pasta de transformações.

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.users_cdf
AS SELECT
  col1 AS userId,
  col2 AS name,
  col3 AS city,
  col4 AS operation,
  col5 AS sequenceNum
FROM (
  VALUES
  -- Initial load.
  (124, "Raul",     "Oaxaca",      "INSERT", 1),
  (123, "Isabel",   "Monterrey",   "INSERT", 1),
  -- New users.
  (125, "Mercedes", "Tijuana",     "INSERT", 2),
  (126, "Lily",     "Cancun",      "INSERT", 2),
  -- Isabel is removed from the system and Mercedes moved to Guadalajara.
  (123, null,       null,          "DELETE", 6),
  (125, "Mercedes", "Guadalajara", "UPDATE", 6),
  -- This batch of updates arrived out of order. The batch at sequenceNum 6 is the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

Processar as atualizações do Tipo 1 de SCD

O SCD Tipo 1 mantém apenas a versão mais recente de cada registo. O exemplo seguinte extrai dados a partir do feed de alterações criado acima e aplica alterações a um alvo de tabela de streaming. Precisas de um pipeline para correr este código. Consulte Desenvolver e depurar pipelines ETL com o Lakeflow Pipelines Editor.

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
  target = "users_current",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)

SQL

CREATE OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_current
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;

Após a execução do exemplo SCD Tipo 1, a tabela alvo contém os seguintes registos:

O utilizador 123 (Isabel) foi eliminado e não aparece. O utilizador 125 (Mercedes) mostra apenas a cidade mais recente (Guadalajara) porque o SCD Tipo 1 sobrespõe valores anteriores. A versão anterior em UPDATE em sequenceNum=5 foi abandonada porque chegou uma atualização posterior em sequenceNum=6.

Após executar o exemplo com o TRUNCATE registo sem comentários, a tabela é limpa em sequenceNum=3. Isto significa que os registos 124 e 126 não estão na tabela, e a tabela alvo final contém apenas o seguinte registo:

Processar atualizações do Tipo 2 de SCD

O SCD Tipo 2 preserva um histórico completo das alterações criando novas linhas para cada versão de um registo, com __START_AT colunas e __END_AT a indicar quando cada versão esteve ativa.

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

Após executar o exemplo SCD Tipo 2, a tabela alvo contém os seguintes registos:

A tabela preserva a história completa. O Utilizador 123 tem duas versões (terminadas na sequência 6 quando eliminadas). Utilizador 125 tem três versões que mostram alterações de cidade. Os registos com __END_AT = null estão atualmente ativos.

Rastreie um subconjunto de colunas com SCD Tipo 2

Por defeito, o SCD Tipo 2 cria uma nova versão sempre que o valor de qualquer coluna muda. Pode especificar um subconjunto de colunas a acompanhar, para que as alterações noutras colunas atualizem a versão atual em vez de gerar um novo registo de histórico.

O exemplo seguinte exclui a city coluna do registo histórico:

Python

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT
  (city)

Como city as alterações não são acompanhadas, as atualizações da cidade sobrescrevem a linha atual em vez de criar uma nova versão. A tabela alvo contém os seguintes registos:

Exemplos de CD AUTOMÁTICO A PARTIR DE SNAPSHOT

As secções seguintes fornecem exemplos da utilização de AUTO CDC FROM SNAPSHOT para processar instantâneos em tabelas alvo SCD Tipo 1 ou Tipo 2. Para contexto sobre quando usar esta API, veja Alterar captura de dados e instantâneos.

Os exemplos nesta secção incluem implementações em Python e SQL. Para sintaxe SQL completa, veja CREATE FLOW (pipelines).

Exemplo: Processamento de instantâneos periódicos

Use esta abordagem quando os snapshots chegam regularmente e por ordem. A interface do Python utiliza a ordem de atualização do pipeline para o versionamento. Para SQL, o exemplo utiliza uma tabela de versões separada para identificar quando um novo snapshot está disponível.

Pode ler snapshots de vários tipos de fonte, incluindo tabelas Delta, ficheiros de armazenamento na cloud e ligações JDBC.

Passo 1: Criar dados de exemplo

Crie uma tabela com dados de instantâneo. Execute o seguinte código a partir de um notebook ou do Databricks SQL na explorations pasta do seu pipeline:

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.snapshot (
  userId INT,
  city STRING
);

CREATE TABLE main.cdc_tutorial.snapshot_version (
  version BIGINT
);

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (1, 'Oaxaca'),
  (2, 'Monterrey'),
  (3, 'Tijuana');

INSERT INTO main.cdc_tutorial.snapshot_version VALUES (0);

Passo 2: Executar AUTO CDC A PARTIR DO SNAPSHOT

Precisas de um pipeline para correr o código nesta etapa. Consulte Desenvolver e depurar pipelines ETL com o Lakeflow Pipelines Editor.

Para Python, escolha um tipo de origem para a vista de instantâneo (o código de criação de exemplo gera uma tabela Delta):

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")

from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return (spark.read
    .format("jdbc")
    .option("url", "<jdbc-url>")
    .option("dbtable", "<table-name>")
    .option("user", "<username>")
    .option("password", "<password>")
    .load()
  )

Depois, adicione a tabela de destino e o fluxo. A implementação SQL utiliza a snapshot_version tabela para identificar quando um novo snapshot está disponível:

Python

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.snapshot
)
WITH VERSION (
  SELECT version
  FROM main.cdc_tutorial.snapshot_version
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
)
KEYS (userId)
STORED AS SCD TYPE 2;

Após a primeira execução do pipeline, todos os registos são inseridos como linhas ativas:

Passo 3: Simular um novo snapshot e repetir

Atualize a tabela de origem para simular a chegada de um novo snapshot (execute este código a partir de um notebook ou ficheiro SQL na explorations pasta do seu pipeline):

TRUNCATE TABLE main.cdc_tutorial.snapshot;

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley'),
  (6, 'Kings Canyon');

UPDATE main.cdc_tutorial.snapshot_version SET version = 1;

Executar o pipeline novamente. AUTO CDC FROM SNAPSHOT compara o novo snapshot com o anterior e deteta que o utilizador 1 foi eliminado, os utilizadores 2 e 3 foram atualizados, e os utilizadores 4 e 6 foram inseridos. Isto gera um feed de alterações e é usado AUTO CDC para criar a tabela de saída.

Após a segunda execução com SCD Tipo 2, a tabela alvo contém os seguintes registos:

O Utilizador 1 foi encerrado (apagado). Os utilizadores 2 e 3 têm cada um duas versões que mostram as alterações das suas cidades. Os utilizadores 4 e 6 foram recém-inseridos.

Após a segunda execução com SCD Tipo 1, a tabela alvo mostra apenas o estado atual:

Exemplo: Processar snapshots usando funções de versão

Use esta abordagem quando precisar de controlo explícito sobre a ordenação de snapshots. Por exemplo, use esta abordagem quando vários snapshots chegam ao mesmo tempo, ou quando chegam fora de ordem. Define como selecionar o próximo snapshot e o seu número de versão. A interface da API processa capturas por ordem crescente de versões.

• Se vários snapshots estiverem armazenados, o seletor de exemplo processa snapshots disponíveis por ordem de versão.
• Se um snapshot chegar fora de ordem (por exemplo, snapshot_3 chega depois de snapshot_4), o seletor de exemplo exclui-o porque a sua versão não é maior do que a última versão confirmada.
• Se não houver novos snapshots, a função version ou a consulta não retorna qualquer resultado e não ocorre qualquer processamento.

Passo 1: Preparar ficheiros de instantâneo

Crie ficheiros CSV contendo dados snapshot e adicione-os a um volume ou local de armazenamento na cloud. Nomeie os ficheiros cronologicamente (por exemplo, snapshot_1.csv, snapshot_2.csv).

Cada ficheiro deve conter colunas para userId e city. Por exemplo:

snapshot_1.csv:

snapshot_2.csv:

Passo 2: Execute AUTO CDC A PARTIR DO SNAPSHOT com uma função de versão

Num pipeline, crie um novo ficheiro Python ou SQL na transformations pasta e adicione o seguinte código. Em seguida, execute o pipeline. Consulte Desenvolver e depurar pipelines ETL com o Lakeflow Pipelines Editor.

Python

from pyspark import pipelines as dp
from typing import Optional, Tuple
from pyspark.sql import DataFrame

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
  snapshot_dir = "/Volumes/main/cdc_tutorial/snapshots/" # or the location you created the sample data

  files = dbutils.fs.ls(snapshot_dir)
  snapshot_files = [f.name for f in files if f.name.startswith("snapshot_") and f.name.endswith(".csv")]

  snapshot_versions = []
  for filename in snapshot_files:
    try:
      version = int(filename.replace("snapshot_", "").replace(".csv", ""))
      snapshot_versions.append(version)
    except ValueError:
      continue

  snapshot_versions.sort()

  if latest_snapshot_version is None:
    if snapshot_versions:
      next_version = snapshot_versions[0]
    else:
      return None
  else:
    next_versions = [v for v in snapshot_versions if v > latest_snapshot_version]
    if next_versions:
      next_version = next_versions[0]
    else:
      return None

  snapshot_path = f"{snapshot_dir}snapshot_{next_version}.csv"
  df = spark.read.format("csv").option("header", True).load(snapshot_path)
  return (df, next_version)


dp.create_streaming_table("main.cdc_tutorial.target_versioned")

dp.create_auto_cdc_from_snapshot_flow(
  target = "main.cdc_tutorial.target_versioned",
  source = next_snapshot_and_version,
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE main.cdc_tutorial.target_versioned;

CREATE FLOW target_versioned_snapshot_flow AS
AUTO CDC INTO main.cdc_tutorial.target_versioned
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'snapshot_',
    CAST((SELECT version FROM current_snapshot_version()) AS STRING),
    '.csv'
  )
)
WITH VERSION (
  WITH snapshot_files AS (
    SELECT
      CAST(REGEXP_EXTRACT(path, 'snapshot_([0-9]+)[.]csv$', 1) AS BIGINT) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/snapshots/')
    WHERE path RLIKE 'snapshot_[0-9]+[.]csv$'
  )
  SELECT version
  FROM snapshot_files
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY version
  LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;

Após o processamento de snapshot_1.csv, a tabela de destino contém os seguintes registos:

Após o processamento de snapshot_2.csv, a tabela de destino contém os seguintes registos:

Passo 3: Adicionar novos snapshots

Adicione um novo ficheiro CSV à localização de armazenamento com dados modificados (por exemplo, valores de cidade alterados, novas linhas ou linhas removidas). Depois executa novamente o pipeline para processar o novo snapshot.

Limitações

• A coluna de sequenciação deve ser um tipo de dado ordenável. NULL Os valores de sequenciação não são suportados.
• Para obter dados em fluxo a partir do destino de um processo AUTO CDC, leia o respetivo fluxo de alterações. Para mais detalhes, consulte Ler um feed de dados de alteração a partir de uma tabela alvo AUTO CDC.

Recursos adicionais

• Captação de Dados de Alterações e snapshots: Aprenda sobre conceitos de CDC, snapshots e tipos de SCD.
• Replicar uma tabela RDBMS externa usando AUTO CDC: Aprenda a realizar a hidratação inicial com um once fluxo e depois continue a processar as alterações.
• Retropreenchimento de dados históricos com pipelines: Aprenda como retropreencher destinos só de acréscimo e AUTO CDC SCD Tipo 1.
• Tópicos avançados do AUTO CDC: Aprenda sobre operações de alteração nos alvos do AUTO CDC, leitura de feeds de dados de alteração e métricas de processamento.
• Rebobinar e reproduzir um pipeline: Aprenda a restaurar um alvo AUTO CDC para um ponto anterior no tempo e a reprocessar as alterações.
• Tutorial: Criar um pipeline de ETL usando a captura de dados de alteração