criar_fluxo_automático_cdc

A função create_auto_cdc_flow() cria um fluxo que usa a funcionalidade de captura de dados de alterações (CDC) de LakeFlow Pipelines para processar dados de origem de um feed de dados de alterações (CDF).

nota

Esta função substitui a função anterior apply_changes(). As duas funções têm a mesma assinatura. A Databricks recomenda atualizar para usar o novo nome.

importante

Você deve declarar uma tabela de transmissão de destino para aplicar as alterações. Opcionalmente, você pode especificar o esquema para sua tabela de destino. Ao especificar o esquema da tabela de destino create_auto_cdc_flow() , você deve incluir as colunas __START_AT e __END_AT com o mesmo tipo de dados dos campos sequence_by .

Para criar a tabela de destino necessária, você pode usar a função create_streaming_table() na interface Python do pipeline.

Sintaxe

Python

from pyspark import pipelines as dp

dp.create_auto_cdc_flow(
  target = "<target-table>",
  source = "<data-source>",
  keys = ["key1", "key2", "keyN"],
  sequence_by = "<sequence-column>",
  system_sequence_by = None, # optional
  ignore_null_updates = False, # optional
  ignore_null_updates_column_list = None, # optional
  ignore_null_updates_except_column_list = None, # optional
  columns_to_update = None, # optional
  apply_as_deletes = None, # optional
  apply_as_truncates = None, # optional
  column_list = None, # optional
  except_column_list = None, # optional
  stored_as_scd_type = "1", # optional
  track_history_column_list = None, # optional
  track_history_except_column_list = None, # optional
  name = None, # optional
  once = False # optional
)

Para o processamento create_auto_cdc_flow , o comportamento default para eventos INSERT e UPDATE é inserir eventos CDC da origem: atualizar todas as linhas na tabela de destino que correspondem à(s) key(s) especificada(s) ou inserir uma nova linha quando um registro correspondente não existir na tabela de destino. O tratamento de eventos DELETE pode ser especificado com o parâmetro apply_as_deletes .

Para saber mais sobre o processamento CDC com um feed de alterações, consulte As APIs AUTO CDC : Simplifique a captura de dados de alterações (CDC) com pipeline. Para um exemplo de utilização da função create_auto_cdc_flow() , consulte os exemplos AUTO CDC.

Parâmetros

Parâmetro	Tipo	Descrição
target	str	Obrigatório. O nome da tabela a ser atualizada. Você pode usar a função create_streaming_table() para criar a tabela de destino antes de executar a função create_auto_cdc_flow() .
source	str	Obrigatório. A fonte de dados contendo registros CDC .
keys	list	Obrigatório. A coluna ou combinação de colunas que identifica exclusivamente uma linha nos dados de origem. Isso é usado para identificar quais eventos do CDC se aplicam a registros específicos na tabela de destino. Você pode especificar: Uma lista de strings: ["userId", "orderId"] Uma lista de funções Spark SQL col() : [col("userId"), col("orderId")]. Argumentos para funções col() não podem incluir qualificadores. Por exemplo, você pode usar col(userId), mas não pode usar col(source.userId).
sequence_by	str, col() ou struct()	Obrigatório. Os nomes das colunas que especificam a ordem lógica dos eventos de CDC nos dados de origem. LakeFlow Pipelines usam esse sequenciamento para manipular eventos de alteração que chegam fora de ordem. A coluna especificada deve ser um tipo de dados classificável. Você pode especificar um dos seguintes: Uma string: "sequenceNum" Uma função Spark SQL col() : col("sequenceNum"). Argumentos para funções col() não podem incluir qualificadores. Por exemplo, você pode usar col(userId), mas não pode usar col(source.userId). Uma struct() que combina múltiplas colunas para desempate: struct("timestamp_col", "id_col"). A ordenação é feita pelo primeiro campo struct, depois pelo segundo campo, se houver empate, e assim sucessivamente.
system_sequence_by	str ou col()	A coluna que especifica o tempo do sistema em que cada evento CDC é conhecido pelo sistema. Usado com stored_as_scd_type="bitemporal" para rastrear alterações entre o tempo de negócios (sequence_by) e o tempo do sistema. A coluna especificada deve ser um tipo de dados classificável. Bitemporal AUTO CDC está em Beta. Consulte Bitemporal AUTO CDC. Este parâmetro é opcional e aplica-se apenas a tabelas bitemporais.
ignore_null_updates	bool	Controla como os valores null são tratados em atualizações CDC recebidas. Quando ignore_null_updates for True, as colunas null em uma atualização recebida são ignoradas; o valor existente na linha de destino é preservado. Isso também se aplica a colunas aninhadas com valores null. Quando ignore_null_updates for False, as colunas null em uma atualização recebida sobrescrevem os valores existentes no destino. O default é False. Defina como True quando os eventos de origem incluem apenas as colunas que foram alteradas, para que as colunas inalteradas não sejam sobrescritas com null. O padrão é False. Para um controle mais preciso sobre quais colunas ignoram valores null, use ignore_null_updates_column_list ou ignore_null_updates_except_column_list.
ignore_null_updates_column_list	list	Um subconjunto de colunas para o qual os valores null em um registro de alteração de entrada são ignorados, de modo que cada uma dessas colunas mantém seu valor existente no destino. Colunas fora da lista aplicam valores explícitos null. Use este parâmetro para aplicar atualizações parciais quando sua origem envia apenas as colunas que foram alteradas. Equivalente à cláusula IGNORE NULL UPDATES ON columnList do SQL. Use ignore_null_updates_column_list ou ignore_null_updates_except_column_list, não ambos.
ignore_null_updates_except_column_list	list	Um subconjunto de colunas que aplicam valores null explícitos. Todas as outras colunas ignoram os valores null em um registro de alteração de entrada e mantêm seu valor existente no destino. Equivalente à cláusula SQL IGNORE NULL UPDATES ON * EXCEPT (...). Use ignore_null_updates_column_list ou ignore_null_updates_except_column_list, não ambos.
columns_to_update	str ou col()	O nome de uma coluna de origem que contém, para cada registro de alteração, o conjunto de colunas a ser atualizado como um array de strings de nomes de colunas (array<string>). As colunas que não estão no array mantêm seus valores de destino existentes, enquanto as colunas listadas são gravadas da origem, incluindo valores null explícitos. Use este parâmetro quando cada registro de alteração atualizar um conjunto diferente de colunas e for necessário aplicar valores null explícitos. Equivalente à cláusula SQL COLUMNS TO UPDATE. Você não pode definir columns_to_update juntamente com ignore_null_updates, ignore_null_updates_column_list ou ignore_null_updates_except_column_list. columns_to_update não é compatível com tabelas bitemporais.
apply_as_deletes	str ou expr()	Especifica quando um evento CDC deve ser tratado como um DELETE em vez de um upsert. Você pode especificar: Uma string: "Operation = 'DELETE'" Uma função Spark SQL expr(): expr("Operation = 'DELETE'") Para lidar com dados fora de ordem, a linha excluída é temporariamente retida como uma marca de exclusão na tabela Delta subjacente, e uma view é criada no metastore que filtra essas marcas de exclusão. O intervalo de retenção padrão é de dois dias e pode ser configurado com a propriedade da tabela pipelines.cdc.tombstoneGCThresholdInSeconds . Se você usar o Auto Loader como fonte para seu pipeline CDC, o Auto Loader não garante a ordem de processamento dos arquivos. Para obter detalhes, consulte Lidar com dados fora de ordem. Defina pipelines.cdc.tombstoneGCThresholdInSeconds com um valor que exceda o atraso máximo esperado entre a chegada do evento e a execução do pipeline. Isso garante que os registros de exclusão sejam mantidos por tempo suficiente para lidar corretamente com eventos de exclusão que chegam atrasados ou fora de ordem.
apply_as_truncates	str ou expr()	Especifica quando um evento de CDC deve ser tratado como uma tabela completa TRUNCATE. Você pode especificar: Uma string: "Operation = 'TRUNCATE'" Uma função Spark SQL expr(): expr("Operation = 'TRUNCATE'") Como esta cláusula aciona um truncamento completo da tabela de destino, ela deve ser usada somente para casos de uso específicos que exigem essa funcionalidade. O parâmetro apply_as_truncates é suportado apenas para SCD tipo 1. O SCD tipo 2 não suporta operações de truncamento.
column_list ou except_column_list	list	Um subconjunto de colunas a incluir na tabela de destino. Use column_list para especificar a lista completa de colunas a incluir. Use except_column_list para especificar as colunas a serem excluídas. Você pode declarar o valor como uma lista de strings ou como funções Spark SQL col(): column_list = ["userId", "name", "city"] column_list = [col("userId"), col("name"), col("city")] except_column_list = ["operation", "sequenceNum"] except_column_list = [col("operation"), col("sequenceNum") Argumentos para funções col() não podem incluir qualificadores. Por exemplo, você pode usar col(userId), mas não pode usar col(source.userId). O default é incluir todas as colunas na tabela de destino quando nenhum argumento column_list ou except_column_list é passado para a função.
stored_as_scd_type	str ou int	Se os registros devem ser armazenados como SCD tipo 1, SCD tipo 2 ou bitemporal. Defina como 1 para SCD tipo 1, 2 para SCD tipo 2, ou "bitemporal" para rastrear alterações no tempo de negócios e no tempo de sistema. Bitemporal requer system_sequence_by e está em Beta. Consulte Bitemporal AUTO CDC. O default é SCD tipo 1.
track_history_column_list ou track_history_except_column_list	list	Um subconjunto de colunas de saída a serem rastreadas para histórico na tabela de destino. Use track_history_column_list para especificar a lista completa de colunas a serem rastreadas. Use track_history_except_column_list para especificar as colunas a serem excluídas do acompanhamento. Você pode declarar qualquer valor como uma lista de strings ou como funções Spark SQL col() : track_history_column_list = ["userId", "name", "city"] track_history_column_list = [col("userId"), col("name"), col("city")] track_history_except_column_list = ["operation", "sequenceNum"] track_history_except_column_list = [col("operation"), col("sequenceNum") Argumentos para funções col() não podem incluir qualificadores. Por exemplo, você pode usar col(userId), mas não pode usar col(source.userId). O default é incluir todas as colunas na tabela de destino quando nenhum argumento track_history_column_list ou track_history_except_column_list é passado para a função.
name	str	O nome do fluxo. Se não for fornecido, o padrão será o mesmo valor de target.
once	bool	Opcionalmente, defina o fluxo como um fluxo único, como um aterro. Usar once=True altera o fluxo de duas maneiras: O valor de retorno. streaming-query. deve ser um lotes DataFrame neste caso, não um transmissão DataFrame. O fluxo é executado uma vez por default. Se o pipeline for atualizado com uma refresh completa, o fluxo ONCE será executado novamente para recriar os dados.

Parâmetro	Tipo	Descrição
target	str	Obrigatório. O nome da tabela a ser atualizada. Você pode usar a função create_streaming_table() para criar a tabela de destino antes de executar a função create_auto_cdc_flow() .
source	str	Obrigatório. A fonte de dados contendo registros CDC .
keys	list	Obrigatório. A coluna ou combinação de colunas que identifica exclusivamente uma linha nos dados de origem. Isso é usado para identificar quais eventos do CDC se aplicam a registros específicos na tabela de destino. Você pode especificar: Uma lista de strings: ["userId", "orderId"] Uma lista de funções Spark SQL col() : [col("userId"), col("orderId")]. Argumentos para funções col() não podem incluir qualificadores. Por exemplo, você pode usar col(userId), mas não pode usar col(source.userId).
sequence_by	str, col() ou struct()	Obrigatório. Os nomes das colunas que especificam a ordem lógica dos eventos de CDC nos dados de origem. LakeFlow Pipelines usam esse sequenciamento para manipular eventos de alteração que chegam fora de ordem. A coluna especificada deve ser um tipo de dados classificável. Você pode especificar um dos seguintes: Uma string: "sequenceNum" Uma função Spark SQL col() : col("sequenceNum"). Argumentos para funções col() não podem incluir qualificadores. Por exemplo, você pode usar col(userId), mas não pode usar col(source.userId). Uma struct() que combina múltiplas colunas para desempate: struct("timestamp_col", "id_col"). A ordenação é feita pelo primeiro campo struct, depois pelo segundo campo, se houver empate, e assim sucessivamente.
system_sequence_by	str ou col()	A coluna que especifica o tempo do sistema em que cada evento CDC é conhecido pelo sistema. Usado com stored_as_scd_type="bitemporal" para rastrear alterações entre o tempo de negócios (sequence_by) e o tempo do sistema. A coluna especificada deve ser um tipo de dados classificável. Bitemporal AUTO CDC está em Beta. Consulte Bitemporal AUTO CDC. Este parâmetro é opcional e aplica-se apenas a tabelas bitemporais.
ignore_null_updates	bool	Controla como os valores null são tratados em atualizações CDC recebidas. Quando ignore_null_updates for True, as colunas null em uma atualização recebida são ignoradas; o valor existente na linha de destino é preservado. Isso também se aplica a colunas aninhadas com valores null. Quando ignore_null_updates for False, as colunas null em uma atualização recebida sobrescrevem os valores existentes no destino. O default é False. Defina como True quando os eventos de origem incluem apenas as colunas que foram alteradas, para que as colunas inalteradas não sejam sobrescritas com null. O padrão é False. Para um controle mais preciso sobre quais colunas ignoram valores null, use ignore_null_updates_column_list ou ignore_null_updates_except_column_list.
ignore_null_updates_column_list	list	Um subconjunto de colunas para o qual os valores null em um registro de alteração de entrada são ignorados, de modo que cada uma dessas colunas mantém seu valor existente no destino. Colunas fora da lista aplicam valores explícitos null. Use este parâmetro para aplicar atualizações parciais quando sua origem envia apenas as colunas que foram alteradas. Equivalente à cláusula IGNORE NULL UPDATES ON columnList do SQL. Use ignore_null_updates_column_list ou ignore_null_updates_except_column_list, não ambos.
ignore_null_updates_except_column_list	list	Um subconjunto de colunas que aplicam valores null explícitos. Todas as outras colunas ignoram os valores null em um registro de alteração de entrada e mantêm seu valor existente no destino. Equivalente à cláusula SQL IGNORE NULL UPDATES ON * EXCEPT (...). Use ignore_null_updates_column_list ou ignore_null_updates_except_column_list, não ambos.
columns_to_update	str ou col()	O nome de uma coluna de origem que contém, para cada registro de alteração, o conjunto de colunas a ser atualizado como um array de strings de nomes de colunas (array<string>). As colunas que não estão no array mantêm seus valores de destino existentes, enquanto as colunas listadas são gravadas da origem, incluindo valores null explícitos. Use este parâmetro quando cada registro de alteração atualizar um conjunto diferente de colunas e for necessário aplicar valores null explícitos. Equivalente à cláusula SQL COLUMNS TO UPDATE. Você não pode definir columns_to_update juntamente com ignore_null_updates, ignore_null_updates_column_list ou ignore_null_updates_except_column_list. columns_to_update não é compatível com tabelas bitemporais.
apply_as_deletes	str ou expr()	Especifica quando um evento CDC deve ser tratado como um DELETE em vez de um upsert. Você pode especificar: Uma string: "Operation = 'DELETE'" Uma função Spark SQL expr(): expr("Operation = 'DELETE'") Para lidar com dados fora de ordem, a linha excluída é temporariamente retida como uma marca de exclusão na tabela Delta subjacente, e uma view é criada no metastore que filtra essas marcas de exclusão. O intervalo de retenção padrão é de dois dias e pode ser configurado com a propriedade da tabela pipelines.cdc.tombstoneGCThresholdInSeconds . Se você usar o Auto Loader como fonte para seu pipeline CDC, o Auto Loader não garante a ordem de processamento dos arquivos. Para obter detalhes, consulte Lidar com dados fora de ordem. Defina pipelines.cdc.tombstoneGCThresholdInSeconds com um valor que exceda o atraso máximo esperado entre a chegada do evento e a execução do pipeline. Isso garante que os registros de exclusão sejam mantidos por tempo suficiente para lidar corretamente com eventos de exclusão que chegam atrasados ou fora de ordem.
apply_as_truncates	str ou expr()	Especifica quando um evento de CDC deve ser tratado como uma tabela completa TRUNCATE. Você pode especificar: Uma string: "Operation = 'TRUNCATE'" Uma função Spark SQL expr(): expr("Operation = 'TRUNCATE'") Como esta cláusula aciona um truncamento completo da tabela de destino, ela deve ser usada somente para casos de uso específicos que exigem essa funcionalidade. O parâmetro apply_as_truncates é suportado apenas para SCD tipo 1. O SCD tipo 2 não suporta operações de truncamento.
column_list ou except_column_list	list	Um subconjunto de colunas a incluir na tabela de destino. Use column_list para especificar a lista completa de colunas a incluir. Use except_column_list para especificar as colunas a serem excluídas. Você pode declarar o valor como uma lista de strings ou como funções Spark SQL col(): column_list = ["userId", "name", "city"] column_list = [col("userId"), col("name"), col("city")] except_column_list = ["operation", "sequenceNum"] except_column_list = [col("operation"), col("sequenceNum") Argumentos para funções col() não podem incluir qualificadores. Por exemplo, você pode usar col(userId), mas não pode usar col(source.userId). O default é incluir todas as colunas na tabela de destino quando nenhum argumento column_list ou except_column_list é passado para a função.
stored_as_scd_type	str ou int	Se os registros devem ser armazenados como SCD tipo 1, SCD tipo 2 ou bitemporal. Defina como 1 para SCD tipo 1, 2 para SCD tipo 2, ou "bitemporal" para rastrear alterações no tempo de negócios e no tempo de sistema. Bitemporal requer system_sequence_by e está em Beta. Consulte Bitemporal AUTO CDC. O default é SCD tipo 1.
track_history_column_list ou track_history_except_column_list	list	Um subconjunto de colunas de saída a serem rastreadas para histórico na tabela de destino. Use track_history_column_list para especificar a lista completa de colunas a serem rastreadas. Use track_history_except_column_list para especificar as colunas a serem excluídas do acompanhamento. Você pode declarar qualquer valor como uma lista de strings ou como funções Spark SQL col() : track_history_column_list = ["userId", "name", "city"] track_history_column_list = [col("userId"), col("name"), col("city")] track_history_except_column_list = ["operation", "sequenceNum"] track_history_except_column_list = [col("operation"), col("sequenceNum") Argumentos para funções col() não podem incluir qualificadores. Por exemplo, você pode usar col(userId), mas não pode usar col(source.userId). O default é incluir todas as colunas na tabela de destino quando nenhum argumento track_history_column_list ou track_history_except_column_list é passado para a função.
name	str	O nome do fluxo. Se não for fornecido, o padrão será o mesmo valor de target.
once	bool	Opcionalmente, defina o fluxo como um fluxo único, como um aterro. Usar once=True altera o fluxo de duas maneiras: O valor de retorno. streaming-query. deve ser um lotes DataFrame neste caso, não um transmissão DataFrame. O fluxo é executado uma vez por default. Se o pipeline for atualizado com uma refresh completa, o fluxo ONCE será executado novamente para recriar os dados.