Flink CDC Transform module - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Parâmetros de regras do Transform

O módulo Transform permite a manipulação direta de colunas de dados. Exclua ou estenda colunas existentes e filtre dados indesejados durante a sincronização de dados. Use os seguintes parâmetros para definir as regras do Transform:

Parâmetro	Descrição	Obrigatório	Observações
`source-table`	Especifique a tabela upstream para a transformação.	Sim	Aceita expressões regulares.
`projection`	Especifique a regra de projeção para a tabela upstream. Isso define todas as colunas de dados após a transformação.	Não	A sintaxe é semelhante à da instrução SELECT do SQL. Se você não especificar esse parâmetro, o sistema não adiciona colunas; não exclua colunas. Para mais informações, consulte Data screening. Para obter detalhes sobre as funções integradas disponíveis, consulte Flink CDC built-in functions.
`filter`	Regra de filtragem de linhas.	Não	A sintaxe é semelhante à da instrução WHERE do SQL. Se omitido, o sistema não filtrará nenhuma linha.
`primary-keys`	Especifique as colunas de chave primária após a transformação.	Não	Se você não especificar esse parâmetro, o sistema manterá a definição de chave primária do schema original. Separe as chaves primárias com vírgulas (`,`). Importante Por padrão, não exclua colunas de chave primária do upstream. Adicione o parâmetro `transform.allow.trimming.pk-columns` à configuração `pipeline:` para reduzir o número de colunas de chave primária do upstream. Ao definir colunas de chave primária personalizadas, certifique-se de que os dados recebidos da tabela upstream estejam em conformidade com a restrição PRIMARY KEY. Para evitar problemas de desordem de dados durante gravações entre partições, recomendamos incluir as colunas de chave primária da tabela upstream nas colunas personalizadas.
`partition-keys`	Especifique a lista de colunas de chave de partição após a transformação.	Não	Caso você não especifique esse parâmetro, o sistema manterá a definição de chave de partição do schema original. Separe as chaves de partição com vírgulas (`,`). Importante Ao definir colunas de chave de partição personalizadas, certifique-se de que os dados recebidos da tabela upstream estejam em conformidade com a restrição PRIMARY KEY para evitar problemas de desordem de dados durante gravações entre partições.
`table-options`	Transmite informações adicionais de configuração para o sink.	Não	Uma lista de propriedades opcionais, como o número de buckets para um sink Paimon ou comentários. Separe os itens de configuração com vírgulas (`,`). Separe a chave e o valor de cada item com um sinal de igual (`=`). Exemplo de configuração: `key1=value1,key2=value2`
`description`	Descrição da regra do Transform.	Não	Nenhuma.
`converter-after-transform`	Conversor que executa processamento adicional nos dados após a conclusão da transformação.	Não	Para mais informações, consulte Converter after Transform.

Observações

Modifique uma instrução no módulo Transform e reinicie o job a partir de um estado inicial sem estado (stateless).
Em geral, as instruções projection e filter não precisam estar entre aspas.
```
transform:
  - projection: a, b, c
    # Equivalent to
  - projection: "a, b, c"
```
No entanto, se uma expressão de projeção começar com um caractere especial, como * ou ', o sistema pode não analisar a expressão inteira como um literal de string YAML válido. Nesse caso, envolva manualmente toda a expressão entre aspas simples (') ou duplas ("), ou use uma barra invertida (\) para escapar o caractere.
```
transform:
  - projection: *, 42      # Not valid YAML
  - projection: '*, 42'    # OK
  - projection: \*, 42     # OK  
```
A partir do Ververica Runtime (VVR) 11.6, você pode limpar as configurações de chave primária inferidas do upstream e converter a tabela em uma tabela sem chave primária. Use a seguinte sintaxe:
```
transform:
  - source-table: db.tbl
    primary-keys: '' # Clears the PRIMARY KEY setting for the db.tbl table
```

Seleção de dados

O módulo Transform usa uma sintaxe semelhante à do SQL para definir regras de seleção (projeção) de dados. Essas regras permitem que você selecione colunas específicas, além de adicionar colunas calculadas e de metadados.

Poda de colunas

Para extrair colunas específicas da tabela source e sincronizá-las com o downstream, especifique as colunas desejadas na regra de projeção. O sistema não envia as colunas não especificadas para o downstream.

transform:
  - source-table: db.tbl
    projection: col_1, col_3, col_4 # col_2 is pruned

Importante

A poda de colunas pode fazer com que os schemas das tabelas upstream e downstream fiquem inconsistentes caso o schema da tabela upstream sofra alterações.

Caracteres curinga

Para enviar todas as colunas da tabela source, incluindo quaisquer novas colunas adicionadas posteriormente, para o downstream sem modificações, use o caractere curinga asterisco (*) na regra de projeção.

Nota

Se uma regra de projeção não usar um caractere curinga (*), o schema resultante será fixo. O schema permanece consistente com a versão especificada na regra de projeção, e o sistema não pode sincronizar a evolução do schema.

Por exemplo, *, 'extras' AS extras indica que o sistema adiciona uma coluna extra após as colunas do schema upstream. O sistema ainda sincroniza a evolução do schema upstream com o downstream.

transform:
  - source-table: db.tbl
    projection: \*, 'extras' AS extras

Colunas calculadas

Use a sintaxe <Expression> AS <ColName> em uma regra de projeção para adicionar uma coluna calculada. O sistema avalia a expressão para cada registro do upstream e preenche a coluna correspondente com o resultado.

Nota

A expressão de uma coluna calculada não pode referenciar o valor de outra coluna calculada, mesmo que a coluna referenciada apareça antes dela. Por exemplo, a, b AS c, c AS d não é uma expressão válida.

Por exemplo, quando a tabela upstream db.tbl envia o registro de dados [+I, id = 1], o sistema transforma o registro na linha de dados [+I, id = 1, inc_id = 2] e, em seguida, o envia para o downstream.

transform:
  - source-table: db.tbl
    projection: id, id + 1 AS inc_id

Colunas de metadados

Ao escrever regras de projeção, use as seguintes colunas de metadados predefinidas da mesma forma que as colunas de dados regulares:

Importante

Não defina colunas de dados regulares com os mesmos nomes das colunas de metadados.

As seguintes colunas de metadados se aplicam a todos os conectores.

Nome da coluna de metadados	Tipo de dados	Descrição
`__namespace_name__`	String	O nome do namespace da tabela source para este registro de alteração.
`__schema_name__`	String	O nome do schema da tabela source para este registro de alteração.
`__table_name__`	String	O nome da tabela source para este registro de alteração.
`__data_event_type__`	String	O tipo de operação para este registro de alteração (`+I`, `-U`, `+U` ou `-D`). Importante Um evento CDC sempre empacota os registros Update Before e Update After de uma atualização em um único evento. Portanto, o conteúdo de `__data_event_type__` é `-U` e `+U`, respectivamente, dentro do mesmo evento de atualização. Não o use como chave primária.

Por exemplo, grave o nome totalmente qualificado da tabela upstream em uma coluna calculada e envie-o para o downstream.

transform:
  - source-table: \.*.\.*
    projection: \*, __namespace_name__ || __schema_name__ || __table_name__ AS identifier

A tabela a seguir mostra as relações de mapeamento entre os conectores de banco de dados e os nomes de namespace, schema e tabela.

Tipo de banco de dados	Nome do namespace	Nome do schema	Nome da tabela
MySQL	-	Database	Table
Kafka	-	-	Topic
SLS	-	Project	LogStore
MongoDB	-	Database	Collection
Paimon	-	Database	Table
Hologres	-	Schema	Table
StarRocks	-	Database	Table
Doris	-	Database	Table
Postgres	Database Nota Isso é válido apenas quando o parâmetro `table-id.include-database` está ativado.	Schema	Table

Filtragem de dados

O módulo Transform usa uma sintaxe semelhante à do SQL para definir regras de filtragem de linhas.

A regra de filtro deve ser uma expressão avaliada como um valor BOOLEAN. Ela pode referenciar qualquer coluna da tabela source e quaisquer colunas calculadas.

Se um registro de alteração corresponder a uma regra do Transform com um filtro e a expressão do filtro for avaliada como FALSE, o sistema não enviará a linha de dados para o downstream.

Nota

Se você usar uma coluna calculada em uma regra de projeção para sobrescrever uma coluna upstream existente, a expressão do filtro referenciará a coluna calculada.

Por exemplo, a seguinte regra do Transform é válida:

transform:
  - source-table: db.tbl
    projection: CAST(id AS VARCHAR) AS id
    filter: CHAR_LENGTH(id) > 5

O id referenciado na expressão filter é a coluna calculada convertida para o tipo VARCHAR.

Correspondência de padrões de string

Use LIKE e NOT LIKE nas expressões de filtro para corresponder a padrões de string. A implementação usa a semântica find() de regex do Java, que difere do LIKE padrão do SQL de duas maneiras importantes:

• O padrão é uma expressão regular do Java. % e _ são tratados como literais de regex, e não como caracteres curinga.

• A correspondência usa a semântica de substring. O padrão precisa aparecer apenas em alguma posição na string para retornar TRUE; não é necessária a correspondência da string inteira.

Exemplo: Filtre registros cuja coluna code comece com HSRS.

O padrão a seguir gera uma correspondência incorreta — ABC_HSRS123 também é correspondido:

transform:
  - source-table: db.orders
    filter: code LIKE 'HSRS.*'

O padrão correto usa a âncora de início ^:

transform:
  - source-table: db.orders
    filter: code LIKE '^HSRS'

Os equivalentes para padrões comuns são os seguintes:

Sintaxe SQL	Regex CDC equivalente	Significado
`col LIKE 'HSRS%'`	`col LIKE '^HSRS'`	Começa com HSRS
`col LIKE '%HSRS'`	`col LIKE 'HSRS$'`	Termina com HSRS
`col LIKE '%HSRS%'`	`col LIKE 'HSRS'`	Contém HSRS
`col LIKE 'HSRS'`	`col LIKE '^HSRS$'`	Exatamente igual a HSRS

Para a semântica completa do LIKE e mais equivalentes de regex, consulte Flink CDC built-in functions.

Regras de configuração avançada

Referenciar colunas não podadas e metadados

Exemplo 1: Filtrar com base em uma coluna podada

Suponha que o schema da tabela upstream seja [INT a, INT b, BOOLEAN c]. Para gerar as colunas a e b e manter apenas as linhas onde c é true, use a seguinte regra de configuração:

transform:
  - source-table: ...
    projection: a, b
    filter: c

Exemplo 2: Filtrar com base em colunas de metadados

Use colunas de metadados diretamente como condições de filtro sem defini-las explicitamente no projection. Por exemplo, filtre dados de alteração do tipo DELETE.

transform:
  - source-table: db.tbl
    projection: a, b
    filter: __data_event_type__ <> '-D'

Sobrescrever colunas existentes

No projection, defina um campo com o mesmo nome de uma coluna upstream para sobrescrever o valor ou o tipo dessa coluna. Esse método garante que o sistema sincronize a evolução do schema corretamente.

Exemplo: Forçar conversão de tipo

Suponha que o schema da tabela upstream seja [INT a, INT b, BOOLEAN c]. Para manter os nomes das colunas inalterados, mas forçar a conversão da coluna a para um tipo string, use a seguinte configuração:

transform:
  - source-table: db.tbl
    projection: \*, CAST(a AS VARCHAR) AS a

O schema da tabela downstream se torna [STRING a, INT b, BOOLEAN c]. A nova definição sobrescreve a coluna original a.

Reutilizar colunas calculadas em condições de filtro

Um filter pode referenciar diretamente o alias de uma coluna calculada definida em um projection.

Exemplo: Referenciar um resultado calculado

Defina uma nova coluna d no projection e use-a diretamente no filter.

transform:
  - source-table: db.tbl
    projection: a, b, c, a + b + c AS d
    filter: d > 100

Essa configuração tem o mesmo efeito que a seguinte, mas é mais legível:

transform:
  - source-table: ...
    projection: a, b, c, a + b + c AS d
    filter: a + b + c > 100

Conversor após o Transform

Use o converter-after-transform para processar alterações de dados após a aplicação de todas as regras do Transform. Especifique vários conversores separando-os por vírgulas (,). O sistema aplica os conversores na ordem em que são listados. Os seguintes valores de configuração são suportados:

Nome do conversor	Recurso	Versões suportadas
SOFT_DELETE	Converte alterações de exclusão em inserções.	VVR 8.0.11 e posterior.
FIELD_NAME_LOWER_CASE	Converte todos os nomes de campo da tabela para minúsculas.	VVR 11.1 e posterior.

Soft delete

O conversor SOFT_DELETE funciona com a coluna de metadados __data_event_type__ para implementar exclusões lógicas. Isso significa que o sistema não exclui os dados fisicamente no downstream. Por exemplo, a seguinte configuração do Transform implementa uma exclusão lógica. O sistema converte uma operação de exclusão em uma inserção. Atualize o op_type dos dados correspondentes para -D para indicar a exclusão.

transform: 
  - source-table: db.tbl
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE

O conversor SOFT_DELETE transforma apenas os dados nos quais __data_event_type__ é -D. Para também ignorar os dados nos quais __data_event_type__ é -U, adicione a configuração filter: op_type <> '-U'.

transform: 
  - source-table: db.tbl
    projection: \*, __data_event_type__ AS op_type
    filter: op_type <> '-U'
    converter-after-transform: SOFT_DELETE