Realtime Compute for Apache Flink:Kafka

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_behavior',
  'properties.bootstrap.servers' = 'localhost:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'csv'
)

CREATE TABLE kafka_source (
  -- Read the message topic as the `record_topic` column
  `record_topic` STRING NOT NULL METADATA FROM 'topic' VIRTUAL,
  -- Read the timestamp from the ConsumerRecord as the `ts` column
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  -- Read the message offset as the `record_offset` column
  `record_offset` BIGINT NOT NULL METADATA FROM 'offset' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE kafka_sink (
  -- Write the timestamp from the `ts` column as the ProducerRecord's timestamp to Kafka
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' VIRTUAL,
  ...
) WITH (
  'connector' = 'kafka',
  ...
);

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="username" password="password";'
)

CREATE TABLE KafkaTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `behavior` STRING,
  `ts` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
) WITH (
  'connector' = 'kafka',
  ...
  'properties.security.protocol' = 'SASL_SSL',
  /* SSL configuration */
  /* Path to the truststore for the server's CA certificate. */
  /* Files uploaded using Artifacts are stored in the /flink/usrlib/ directory. */
  'properties.ssl.truststore.location' = '/flink/usrlib/kafka.client.truststore.jks',
  'properties.ssl.truststore.password' = 'test1234',
  /* If client authentication is required, you must also configure the path to the keystore (private key). */
  'properties.ssl.keystore.location' = '/flink/usrlib/kafka.client.keystore.jks',
  'properties.ssl.keystore.password' = 'test1234',
  /* The algorithm used to verify the server hostname. An empty string disables hostname verification. */
  'properties.ssl.endpoint.identification.algorithm' = '',
  /* SASL configuration */
  /* Set the SASL mechanism to SCRAM-SHA-256. */
  'properties.sasl.mechanism' = 'SCRAM-SHA-256',
  /* Configure JAAS. */
  'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'
)

CREATE TEMPORARY TABLE kafka_source (
  ...
) WITH (
  'connector' = 'kafka',
  ...
  -- Consume from the earliest offset.
  'scan.startup.mode' = 'earliest-offset',
  -- Consume from the latest offset.
  'scan.startup.mode' = 'latest-offset',
  -- Consume from the committed offsets of the consumer group "my-group".
  'properties.group.id' = 'my-group',
  'scan.startup.mode' = 'group-offsets',
  'properties.auto.offset.reset' = 'earliest', -- If "my-group" is used for the first time, consumption starts from the earliest offset.
  'properties.auto.offset.reset' = 'latest', -- If "my-group" is used for the first time, consumption starts from the latest offset.
  -- Consume from the specified timestamp in milliseconds: 1655395200000.
  'scan.startup.mode' = 'timestamp',
  'scan.startup.timestamp-millis' = '1655395200000',
  -- Consume from specific offsets.
  'scan.startup.mode' = 'specific-offsets',
  'scan.startup.specific-offsets' = 'partition:0,offset:42;partition:1,offset:300'
);

CREATE TEMPORARY TABLE kafka_source (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

CREATE TEMPORARY TABLE kafka_sink (
  id INT,
  name STRING,
  age INT
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'properties.group.id' = '<yourKafkaConsumerGroupId>',
  'format' = 'csv'
);

INSERT INTO kafka_sink SELECT id, name, age FROM kafka_source;

CREATE TEMPORARY TABLE kafkaTable (
  `offset` INT NOT NULL METADATA,
  `part` BIGINT NOT NULL METADATA FROM 'partition',
  PRIMARY KEY (`part`, `offset`) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.infer-schema.flatten-nested-columns.enable' = 'true'
    -- Optional. Flattens all nested columns.
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `key_id` INT NOT NULL,
  `val_name` VARCHAR(200)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'val_',
  'value.fields-include' = 'EXCEPT_KEY'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka`(
WITH (
  'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable;

CREATE TEMPORARY TABLE kafkaTable (
  `distinct_id` INT NOT NULL,
  `properties` STRING,
  `timestamp` TIMESTAMP_LTZ METADATA,
  `date` AS CAST(`timestamp` AS DATE)
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = '<yourKafkaBrokers>',
  'topic' = 'kafka_evolution_demo',
  'scan.startup.mode' = 'earliest-offset',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields' = 'key_id',
  'key.fields-prefix' = 'key_'
);

CREATE TABLE IF NOT EXISTS hologres.kafka.`sync_kafka` WITH (
   'connector' = 'hologres'
) AS TABLE vvp.`default`.kafkaTable
ADD COLUMN
  `order_id` AS COALESCE(JSON_VALUE(`properties`, '$.order_id'), 'default');
--Use COALESCE to handle null values.

{
  "id": 101,
  "name": "VVP",
  "properties": {
    "owner": "Alibaba Cloud",
    "engine": "Flink"
  }
}

CREATE TEMPORARY TABLE kafka_source (
  id          VARCHAR,
  `name`      VARCHAR,
  properties  ROW<`owner` STRING, engine STRING>
) WITH (
  'connector' = 'kafka',
  'topic' = 'xxx',
  'properties.bootstrap.servers' = 'xxx',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

source:
  type: kafka
  name: Kafka source
  properties.bootstrap.servers: localhost:9092
  topic: ${kafka.topic}

sink:
  type: kafka
  name: Kafka Sink
  properties.bootstrap.servers: localhost:9092

Parâmetro

• Geral

  Parâmetro	Descrição	Obrigatório	Tipo	Padrão	Observações
  type	Tipo de source ou sink.	Sim	String	–	O valor deve ser kafka.
  name	Nome do source ou sink.	Não	String	–	–
  properties.bootstrap.servers	Endereços dos brokers Kafka.	Sim	String	–	O formato é host:port,host:port,host:port, separado por vírgulas (,).
  properties.*	Propriedades de configuração para o cliente Kafka.	Não	String	–	As chaves de propriedade devem ser opções válidas conforme definido na documentação oficial do Apache Kafka para Producer Configs e Consumer Configs. O Realtime Compute for Apache Flink (VVR) remove o prefixo properties. antes de passar os pares chave-valor restantes para o cliente Kafka subjacente. Por exemplo, defina 'properties.allow.auto.create.topics' = 'false' para desativar a criação automática de tópicos.
  key.format	Formato de serialização e desserialização da chave da mensagem Kafka.	Não	String	–	Para o source, apenas o formato json é suportado. Para o sink, os valores válidos são: csv json Nota Esta opção é suportada apenas no Realtime Compute for Apache Flink (VVR) 11.0.0 e versões posteriores.
  value.format	Formato de serialização e desserialização do valor da mensagem Kafka.	Não	String	debezium-json	Para o source, os valores válidos são: debezium-json canal-json json Para o sink, os valores válidos são: debezium-json canal-json canal-protobuf Nota Os formatos debezium-json e canal-json exigem a versão 8.0.10 ou posterior do Realtime Compute for Apache Flink (VVR). O formato json exige a versão 11.0.0 ou posterior do Realtime Compute for Apache Flink (VVR).

• Parâmetros de source

  Parâmetro	Descrição	Obrigatório	Tipo	Padrão	Observações
  topic	O tópico ou tópicos de onde ler.	Não	String	–	Para assinar vários tópicos, separe seus nomes com ponto e vírgula (;), por exemplo, topic-1;topic-2. Nota Especifique este parâmetro ou topic-pattern, mas não ambos.
  topic-pattern	Uma expressão regular que corresponde aos nomes dos tópicos a serem assinados.	Não	String	–	Exemplos: user_event_.*: Corresponde a todos os tópicos com o prefixo user_event_. prod\.logs\..*: Corresponde a tópicos com o prefixo prod.logs. (o caractere . deve ser escapado). Nota Especifique este parâmetro ou topic, mas não ambos.
  properties.group.id	ID do grupo de consumidores.	Não	String	–	Ao especificar um novo ID de grupo de consumidores, você deve definir o parâmetro properties.auto.offset.reset como earliest ou latest para determinar o offset inicial de leitura.
  scan.startup.mode	Offset inicial do consumidor Kafka.	Não	String	group-offsets	Valores válidos: earliest-offset: Inicia a leitura a partir do offset mais antigo disponível. latest-offset: Inicia a leitura a partir do offset mais recente. group-offsets (Valor padrão): Inicia a leitura a partir dos offsets confirmados para o properties.group.id especificado. timestamp: Inicia a leitura a partir do timestamp especificado por scan.startup.timestamp-millis. specific-offsets: Inicia a leitura a partir dos offsets especificados por scan.startup.specific-offsets. Nota Este parâmetro se aplica apenas quando um job inicia sem estado (stateless). Quando um job com estado inicia, ele sempre consome a partir dos offsets armazenados em seu estado.
  scan.startup.specific-offsets	Offset inicial por partição quando scan.startup.mode está definido como specific-offsets.	Não	String	–	Por exemplo, partition:0,offset:42;partition:1,offset:300
  scan.startup.timestamp-millis	O timestamp inicial em milissegundos quando scan.startup.mode está definido como timestamp.	Não	Long	–	A unidade é milissegundos.
  scan.topic-partition-discovery.interval	Intervalo para descoberta dinâmica de novas partições dentro dos tópicos.	Não	Duration	5 minutos	O conector descobre periodicamente e lê novas partições. Ao usar topic-pattern, o conector também descobre novos tópicos que correspondem ao padrão. Para desativar a descoberta, defina este valor como 0 ou menos.
  scan.check.duplicated.group.id	Verifica se o grupo de consumidores especificado por properties.group.id é duplicado.	Não	Boolean	false	Valores válidos: true: Verifica a existência de um grupo de consumidores duplicado antes do início do job. Se uma duplicidade for encontrada, o job falhará. false: Inicia o job sem verificar conflitos.
  schema.inference.strategy	Estratégia de análise de schema.	Não	String	continuous	Valores válidos: continuous: Analisa o schema de cada registro de dados. Se os schemas forem incompatíveis, o sistema infere um schema mais amplo e gera um evento de alteração de schema. static: Executa a análise do schema apenas uma vez no início do job. Os dados são então analisados com base neste schema inicial, e nenhum evento de alteração de schema é gerado. Nota Para obter mais informações sobre a análise de schema, consulte Políticas para análise e evolução de schema. Esta opção de configuração é suportada apenas no Ververica Runtime (VVR) 8.0.11 e versões posteriores.
  scan.max.pre.fetch.records	Número máximo de mensagens consumidas por partição para inferência inicial de schema.	Não	Int	50	Antes que o processamento de dados comece, o sistema pré-busca e consome o número especificado de mensagens recentes de cada partição para inicializar o schema.
  key.fields-prefix	Prefixo para nomes de campos de chave de mensagem para evitar conflitos de nomes.	Não	String	–	Por exemplo, se este parâmetro estiver definido como key_ e a chave da mensagem contiver um campo chamado a, o nome do campo analisado se tornará key_a. Nota O valor de key.fields-prefix não pode ser um prefixo do valor de value.fields-prefix.
  value.fields-prefix	Prefixo para nomes de campos de valor de mensagem para evitar conflitos de nomes.	Não	String	–	Por exemplo, se este parâmetro estiver definido como value_ e o valor da mensagem contiver um campo chamado b, o nome do campo analisado se tornará value_b. Nota O valor de value.fields-prefix não pode ser um prefixo do valor de key.fields-prefix.
  metadata.list	Colunas de metadados passadas para o sink downstream.	Não	String	–	As colunas de metadados disponíveis incluem topic, partition, offset, timestamp, timestamp-type, headers e leader-epoch. Separe os nomes das colunas com vírgulas.
  scan.value.initial-schemas.ddls	Instruções DDL que definem o schema inicial para tabelas específicas.	Não	String	–	Use ponto e vírgula (;) para separar várias instruções DDL. Por exemplo, use CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10)); CREATE TABLE db1.t2 (id BIGINT); para especificar o schema inicial para as tabelas db1.t1 e db1.t2, respectivamente. O schema da tabela definido no DDL deve ser consistente com a tabela sink de destino e estar em conformidade com a sintaxe do Flink SQL. Nota Esta opção de configuração é suportada apenas no Ververica Runtime (VVR) 11.5 e versões posteriores.
  ingestion.ignore-errors	Ignora erros de análise de dados.	Não	Boolean	false	Nota Esta opção de configuração é suportada apenas no Ververica Runtime (VVR) 11.5 e versões posteriores.
  ingestion.error-tolerance.max-count	Número máximo de erros de análise tolerados antes que o job falhe. Tem efeito apenas quando ingestion.ignore-errors é true.	Não	Integer	-1	Este parâmetro se aplica apenas quando ingestion.ignore-errors está definido como true. Um valor de -1 indica tolerância ilimitada, o que significa que exceções de análise não causarão falha no job. Nota Esta opção de configuração é suportada apenas no Ververica Runtime (VVR) 11.5 e versões posteriores.
  scan.duplicate-field.strategy	Define como lidar com nomes de campos duplicados analisados nas partes de chave e valor.	Não	String	EXCEPTION	Valores válidos: EXCEPTION: Lança uma exceção quando existem campos duplicados na chave e no valor. Este é o comportamento padrão no VVR 11.6 e anteriores. PREFER_KEY: Usa o valor do campo de chave quando os campos estão duplicados. PREFER_VALUE: Usa o valor do campo de valor quando os campos estão duplicados. Nota Esta opção de configuração é suportada apenas no Ververica Runtime (VVR) 11.7 e versões posteriores.

  • Parâmetros do formato Debezium JSON

    Parâmetro	Obrigatório	Tipo	Padrão	Descrição
    debezium-json.distributed-tables	Não	Boolean	false	Defina como true se os dados de uma única tabela Debezium JSON estiverem distribuídos em várias partições. Nota Esta opção de configuração é suportada apenas no Ververica Runtime (VVR) 8.0.11 e versões posteriores. Importante Modificar este parâmetro requer uma inicialização sem estado (stateless startup).
    debezium-json.schema-include	Não	Boolean	false	Inclui um schema na mensagem Debezium JSON. Isso corresponde à propriedade value.converter.schemas.enable na configuração do Debezium Kafka Connect. Valores válidos: true: A mensagem Debezium JSON contém um schema. false: A mensagem Debezium JSON não contém um schema.
    debezium-json.ignore-parse-errors	Não	Boolean	false	Valores válidos: true: Ignora linhas que causam uma exceção de análise. false: Lança um erro e o job falha.
    debezium-json.infer-schema.primitive-as-string	Não	Boolean	false	Analisa todos os tipos primitivos como String ao analisar o schema da tabela. Valores válidos: true: Analisa todos os tipos primitivos como String. false: Analisa os tipos com base nas regras padrão.

  • Parâmetros do formato Canal JSON

    Parâmetro	Obrigatório	Tipo	Padrão	Descrição
    canal-json.distributed-tables	Não	Boolean	false	Se os dados de uma única tabela em Canal JSON estiverem distribuídos em várias partições, você deve ativar esta opção. Nota Esta opção de configuração é suportada apenas no Ververica Runtime (VVR) 8.0.11 e versões posteriores. Importante Modificar este parâmetro requer uma inicialização sem estado (stateless startup).
    canal-json.database.include	Não	String	–	Uma expressão regular opcional para filtrar changelogs pelo campo de metadados database nos registros Canal. Apenas registros de bancos de dados correspondentes são processados. A expressão regular é compatível com a classe Pattern do Java.
    canal-json.table.include	Não	String	–	Uma expressão regular opcional para filtrar changelogs pelo campo de metadados table nos registros Canal. Apenas registros de tabelas correspondentes são processados. A expressão regular é compatível com a classe Pattern do Java.
    canal-json.ignore-parse-errors	Não	Boolean	false	Valores válidos: true: Ignora a linha atual se ocorrer uma exceção de análise. false: Lança um erro e o job falha ao iniciar.
    canal-json.infer-schema.primitive-as-string	Não	Boolean	false	Analisa todos os tipos primitivos como String ao analisar o schema da tabela. Valores válidos: true: Analisa todos os tipos primitivos como String. false: Analisa os tipos com base nas regras padrão.
    canal-json.infer-schema.strategy	Não	String	AUTO	Estratégia de análise de schema da tabela. Valores válidos: AUTO: Analisa automaticamente o schema a partir dos dados JSON. Recomendado se os dados não contiverem um campo sqlType, para evitar falhas na análise. SQL_TYPE: Analisa o schema a partir do array sqlType nos dados Canal JSON. Recomendamos definir isso como SQL_TYPE para obter tipos mais precisos se os dados contiverem um campo sqlType. MYSQL_TYPE: Analisa o schema a partir do array mysqlType nos dados Canal JSON. Para obter mais informações sobre as regras de mapeamento de tipo sqlType, consulte Análise de Schema do Canal JSON. Nota Esta configuração é suportada apenas no Ververica Runtime (VVR) 11.1 e versões posteriores. O valor MYSQL_TYPE é suportado no Ververica Runtime (VVR) 11.3 e versões posteriores.
    canal-json.mysql.treat-mysql-timestamp-as-datetime-enabled	Não	Boolean	true	Mapeia o tipo MySQL TIMESTAMP para o tipo CDC TIMESTAMP. true: O tipo MySQL TIMESTAMP é mapeado para o tipo CDC TIMESTAMP. false: O tipo MySQL TIMESTAMP é mapeado para o tipo CDC TIMESTAMP_LTZ.
    canal-json.mysql.treat-tinyint1-as-boolean.enabled	Não	Boolean	true	Ao usar a estratégia de análise MYSQL_TYPE, controla se deve mapear o tipo MySQL TINYINT(1) para o tipo CDC BOOLEAN. true: O tipo MySQL TINYINT(1) é mapeado para o tipo CDC BOOLEAN. false: O tipo MySQL TINYINT(1) é mapeado para o tipo CDC TINYINT(1). Esta opção se aplica apenas quando canal-json.infer-schema.strategy está definido como MYSQL_TYPE.

  • Parâmetros do formato JSON

    Parâmetro	Obrigatório	Tipo	Padrão	Descrição
    json.timestamp-format.standard	Não	String	SQL	O formato de timestamp para dados de entrada e saída. SQL: Analisa timestamps de entrada no formato yyyy-MM-dd HH:mm:ss.s{precision}, como 2020-12-30 12:13:14.123. ISO-8601: Analisa timestamps de entrada no formato yyyy-MM-ddTHH:mm:ss.s{precision}, como 2020-12-30T12:13:14.123.
    json.ignore-parse-errors	Não	Boolean	false	Valores válidos: true: Ignora a linha atual se ocorrer uma exceção de análise. false: Lança um erro e o job falha ao iniciar.
    json.infer-schema.primitive-as-string	Não	Boolean	false	Analisa todos os tipos primitivos como String ao analisar o schema da tabela. Valores válidos: true: Analisa todos os tipos primitivos como String. false: Analisa os tipos com base nas regras padrão.
    json.infer-schema.flatten-nested-columns.enable	Não	Boolean	false	Expande recursivamente colunas aninhadas em dados JSON. Valores válidos: true: Expande recursivamente colunas aninhadas. false: Trata colunas aninhadas como String.
    json.decode.parser-table-id.fields	Não	String	–	Usa os valores de campos JSON especificados para gerar um tableId ao analisar dados no formato JSON. Os valores de vários campos são concatenados por uma vírgula ,. Por exemplo, se os dados JSON forem {"col0":"a", "col1","b", "col2","c"}, o resultado gerado será o seguinte: Configuração tableId col0 a col0,col1 a.b col0,col1,col2 a.b.c
    json.infer-schema.fixed-types	Não	String	–	Ao analisar dados JSON, você pode especificar os tipos de dados para campos específicos. Use uma vírgula , para separar vários campos. Por exemplo, id BIGINT, name VARCHAR(10) especifica que o campo id é do tipo BIGINT e o campo name é do tipo VARCHAR(10). Nota Esta opção de configuração é suportada apenas no Ververica Runtime (VVR) 11.5 e versões posteriores. Ao usar esta configuração com o Ververica Runtime (VVR) versão 11.5, você também deve adicionar a configuração scan.max.pre.fetch.records: 0.
    json.decode.empty-value-as-delete.enabled	Não	Boolean	false	Especifica se deve analisar mensagens tombstone (com valor vazio) em um tópico compactado do Kafka como eventos DELETE. Usado para cenários onde um valor vazio representa semântica de exclusão, como espelhamento de tópicos compactados ou sinais de exclusão CDC. Nota Esta opção de configuração é suportada apenas no Ververica Runtime (VVR) 11.7 e versões posteriores.

• Parâmetros da tabela sink

  Parâmetro	Descrição	Obrigatório	Tipo	Padrão	Observações
  type	Tipo de sink.	Sim	String	–	O valor deve ser kafka.
  name	Nome do sink.	Não	String	–	–
  topic	Nome do tópico Kafka.	Não	String	–	Se este parâmetro for especificado, todos os dados serão gravados neste tópico. Nota Se este parâmetro não for especificado, cada registro será gravado em um tópico nomeado de acordo com seu TableID. O TableID é construído unindo os nomes do banco de dados e da tabela com um ponto (.), por exemplo, databaseName.tableName.
  partition.strategy	Estratégia de gravação de partição Kafka.	Não	String	all-to-zero	Valores válidos: all-to-zero (padrão): Grava todos os dados na Partição 0. hash-by-key: Grava dados em partições com base no valor hash da chave primária. Isso garante que registros com a mesma chave primária sejam gravados na mesma partição, preservando sua ordem.
  sink.tableId-to-topic.mapping	Mapeamento de nomes de tabelas upstream para nomes de tópicos Kafka downstream.	Não	String	–	Separe os mapeamentos com ponto e vírgula (;). Dentro de cada mapeamento, separe o nome da tabela upstream e o nome do tópico Kafka downstream com dois pontos (:). Você pode usar uma expressão regular para o nome da tabela. Para mapear várias tabelas para o mesmo tópico, separe os nomes das tabelas com vírgulas (,). Por exemplo: mydb.mytable1:topic1;mydb.mytable2:topic2. Nota Este parâmetro permite modificar o tópico mapeado enquanto preserva as informações originais do nome da tabela.

  • Parâmetros do formato Debezium JSON

    Parâmetro	Obrigatório	Tipo	Padrão	Descrição
    debezium-json.include-schema.enabled	Não	Boolean	false	Inclui informações de schema nos dados Debezium JSON.
    debezium-json.emit.full-table-id.enabled	Não	Boolean	false	Grava o ID completo da tabela em três partes nos campos de metadados Debezium JSON. Se este parâmetro estiver ativado, o mapeamento será o seguinte: Parte do ID da Tabela CDC Chave Debezium JSON Namespace db Schema schema Table table Se este parâmetro estiver desativado, o mapeamento será o seguinte: Parte do ID da Tabela CDC Chave Debezium JSON Namespace Não mapeado Schema db Table table Nota Este parâmetro é suportado apenas no Ververica Runtime (VVR) 11.6 e versões posteriores.

Inicialização do schema da tabela

O schema de uma tabela inclui colunas e tipos de dados, nomes de banco de dados e tabelas, além de chaves primárias. As seções a seguir descrevem como inicializar cada um desses elementos.

• Informações de coluna e tipo de dados

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Set the initial table schema
  scan.value.initial-schemas.ddls: CREATE TABLE db1.t1 (id BIGINT, name VARCHAR(10)); CREATE TABLE db1.t2 (id BIGINT);

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: host:9092
  topic: test-topic
  value.format: json
  scan.startup.mode: earliest-offset
  # Set specific fields to a fixed type
  json.infer-schema.fixed-types: id BIGINT, name VARCHAR(10)
  scan.max.pre.fetch.records: 0

• Informações de banco de dados e tabela

  • Nos formatos Canal JSON e Debezium JSON, o conector extrai as informações da tabela, incluindo nome do banco de dados e da tabela, diretamente de cada mensagem.

  • No formato JSON, por padrão, as informações da tabela contêm apenas o nome da tabela, que corresponde ao nome do tópico onde os dados residem. Caso seus dados incluam informações de banco de dados e tabela, utilize o parâmetro json.infer-schema.fixed-types para indicar os campos que armazenam essas informações. O sistema mapeará esses campos para os nomes do banco de dados e da tabela. Veja um exemplo de configuração:

    source:
      type: kafka
      name: Kafka Source
      properties.bootstrap.servers: host:9092
      topic: test-topic
      value.format: json
      scan.startup.mode: earliest-offset
      # Use the value of the col1 field as the database name and the value of the col2 field as the table name
      json.decode.parser-table-id.fields: col1,col2

    Com essa configuração, o conector envia cada registro para uma tabela cujo nome do banco de dados corresponde ao valor do campo col1 e o nome da tabela corresponde ao valor do campo col2.

• Informações de chave primária

  • No formato Canal JSON, o campo pkNames nos dados JSON define a chave primária da tabela.

  • Nos formatos Debezium JSON e JSON, os dados não contêm informações de chave primária. Adicione chaves primárias manualmente às tabelas usando regras de transform:

    transform:
      - source-table: \.*.\.*
        projection: \*
        primary-keys: key1, key2

Análise e evolução de schema

Após a inicialização do schema da tabela, se schema.inference.strategy estiver definido como static, o conector Kafka analisa o valor de cada mensagem com base no schema inicial da tabela e não gera eventos de alteração de schema. Quando schema.inference.strategy está configurado como continuous, o conector examina o valor de cada mensagem Kafka, identifica suas colunas físicas e compara o schema resultante com o schema mantido atualmente. Se houver inconsistência entre os schemas, o conector tenta mesclá-los e gera um evento correspondente de alteração de schema da tabela. As regras de mesclagem funcionam da seguinte forma:

• Caso as colunas físicas analisadas contenham campos ausentes no schema atual, o sistema adiciona esses campos ao schema e gera um evento para incluí-los como colunas anuláveis.

• Se as colunas físicas analisadas não incluírem campos presentes no schema atual, tais campos permanecem e seus valores são preenchidos com NULL. Nenhum evento de exclusão de coluna é gerado.

• Colunas com o mesmo nome recebem o seguinte tratamento:

  • Quando as colunas possuem o mesmo tipo de dados, mas precisões diferentes, o tipo com maior precisão prevalece e um evento de alteração de tipo de coluna é emitido.

  • Se as colunas apresentarem tipos de dados distintos, o sistema localiza o menor tipo pai comum na árvore hierárquica de tipos abaixo. Esse tipo pai comum passa a ser usado para a coluna, gerando-se um evento de alteração de tipo.

    

• Políticas de evolução de schema suportadas:

  • Adição de coluna: O conector insere a nova coluna ao final do schema e sincroniza seus dados, definindo-a como anulável.

  • Remoção de coluna: Não há geração de evento de exclusão. Em vez disso, os dados subsequentes para essa coluna são preenchidos com NULL.

  • Renomeação de coluna: O conector interpreta essa ação como a remoção da coluna antiga e a adição de uma nova. A nova coluna vai para o final do schema, enquanto os valores da coluna original recebem NULL.

  • Alteração de tipo de coluna:

    • Para sinks downstream compatíveis com mudanças de tipo, um job Flink CDC consegue lidar com alterações (por exemplo, de INT para BIGINT) desde que o sink esteja configurado para processá-las. Essa capacidade depende das regras de alteração de tipo suportadas pelo sink específico. Consulte a documentação do seu sink para verificar as regras aplicáveis.

    • Para sinks downstream que não aceitam alterações de tipo de coluna, como o Hologres, utilize o recurso de Ampliação de tipo. Ele cria uma tabela com tipos de dados mais amplos no sink downstream durante o início do job. Assim, quando ocorre uma mudança de tipo, o sistema tolera a alteração contanto que o novo tipo caiba dentro do tipo mais amplo definido no sink.

• Alterações de schema não suportadas:

  • Modificações em restrições, como chaves primárias ou índices.

  • Mudança de uma coluna de NOT NULL para NULLABLE.

• Análise de schema Canal JSON

  Dados Canal JSON podem conter um campo opcional sqlType, que registra informações precisas de tipo para as colunas de dados. Para obter um schema mais preciso, defina canal-json.infer-schema.strategy como SQL_TYPE e utilize os tipos provenientes do campo sqlType. Os mapeamentos de tipo são:

  Tipo JDBC	Código do tipo	Tipo CDC
  BIT	-7	BOOLEAN
  BOOLEAN	16
  TINYINT	-6	TINYINT
  SMALLINT	5	SMALLINT
  INTEGER	4	INT
  BIGINT	-5	BIGINT
  DECIMAL	3	DECIMAL(38,18)
  NUMERIC	2
  REAL	7	FLOAT
  FLOAT	6
  DOUBLE	8	DOUBLE
  BINARY	-2	BYTES
  VARBINARY	-3
  LONGVARBINARY	-4
  BLOB	2004
  DATE	91	DATE
  TIME	92	TIME
  TIMESTAMP	93	TIMESTAMP
  CHAR	1	STRING
  VARCHAR	12
  LONGVARCHAR	-1
  Outros tipos de dados