Conector YAML do Hologres para ingestão de dados - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Este tópico descreve como utilizar o conector do Hologres para sincronizar dados em uma implantação de ingestão de dados YAML.

Contexto

Hologres é um mecanismo de data warehouse em tempo real que suporta escritas, atualizações e análises em escala. O Hologres é compatível com o protocolo PostgreSQL e aceita SQL padrão. Ele oferece suporte a processamento analítico online (OLAP) e consultas ad hoc em petabytes de dados, atendendo requisições com alta concorrência e baixa latência. O Hologres integra-se ao MaxCompute, Realtime Compute for Apache Flink e DataWorks para fornecer uma solução completa de data warehouse online e offline. A tabela a seguir detalha as capacidades do conector YAML do Hologres.

Item	Descrição
Tipo de tabela	Sink
Modo de execução	Modos streaming e batch
Formato de dados	N/A
Métrica	numRecordsOut numRecordsOutPerSecond Nota Para mais detalhes, consulte Metrics.
Tipo de API	YAML
Atualizações ou exclusões em uma tabela sink	Suportado

Recursos

Recurso	Descrição
Full database synchronization	Sincroniza dados completos e incrementais em tempo real de um banco de dados inteiro ou de múltiplas tabelas para as tabelas sink correspondentes.
Schema change synchronization	Replica alterações de esquema (como adição, remoção ou renomeação de colunas) das tabelas de origem para as tabelas sink correspondentes em tempo real.
Sharded database and table synchronization	Utilize expressões regulares para corresponder tabelas de origem em vários bancos de dados fragmentados por nome. Os dados dessas tabelas são mesclados e sincronizados para tabelas sink downstream com nomes correspondentes.
Writing to a partitioned table	Grava dados de uma tabela upstream em uma tabela particionada do Hologres.
Data type mapping	Mapeia tipos de dados upstream para tipos de dados mais abrangentes do Hologres usando diversas estratégias.

Sintaxe

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}

Parâmetros

Parâmetro	Descrição	Tipo	Obrigatório	Padrão	Observações
type	O tipo do sink.	String	Sim	Nenhum	O valor deve ser `hologres`.
name	O nome do sink.	String	Não	Nenhum	N/A.
dbname	O nome do banco de dados.	String	Sim	Nenhum	N/A.
username	O nome de usuário para acesso ao banco de dados. Utilize o AccessKey ID da sua conta Alibaba Cloud.	String	Sim	Nenhum	Para mais informações, consulte How do I view the AccessKey ID and AccessKey secret? Importante Para evitar a exposição do seu AccessKey, especifique seus valores utilizando variáveis. Para mais informações, consulte Project variables.
password	A senha para acesso ao banco de dados. Utilize o AccessKey Secret da sua conta Alibaba Cloud.	String	Sim	Nenhum
endpoint	O endpoint do service Hologres.	String	Sim	Nenhum	Para mais informações, consulte Access endpoints.
jdbcRetryCount	Número de tentativas para operações de escrita e consulta quando ocorre falha na conexão.	Integer	Não	10	N/A.
jdbcRetrySleepInitMs	Tempo de espera fixo para cada tentativa de repetição.	Long	Não	1000	Unidade: milissegundos. O tempo real de espera para uma nova tentativa é calculado pela seguinte fórmula: `jdbcRetrySleepInitMs + retry * jdbcRetrySleepStepMs`.
jdbcRetrySleepStepMs	Tempo de espera incremental para cada tentativa de repetição.	Long	Não	5000	Unidade: milissegundos. O tempo real de espera para uma nova tentativa é calculado pela seguinte fórmula: `jdbcRetrySleepInitMs + retry * jdbcRetrySleepStepMs`.
jdbcConnectionMaxIdleMs	Tempo máximo de ociosidade para uma conexão JDBC.	Long	Não	60000	Unidade: milissegundos. Se uma conexão permanecer ociosa por mais tempo que essa duração, ela será desconectada e liberada.
jdbcMetaCacheTTL	Tempo de expiração para informações de TableSchema armazenadas em cache localmente.	Long	Não	60000	Unidade: milissegundos.
jdbcMetaAutoRefreshFactor	Fator que determina o gatilho de atualização do cache. O sistema atualiza automaticamente o cache se o tempo de vida restante for inferior ao tempo do gatilho.	Integer	Não	4	O tempo de vida restante do cache é calculado da seguinte forma: Tempo de vida restante = Tempo de expiração do cache - Tempo em que o cache esteve ativo. Após uma atualização automática, o tempo ativo do cache é redefinido para 0. O tempo do gatilho é calculado usando a fórmula: `jdbcMetaCacheTTL` / `jdbcMetaAutoRefreshFactor`.
mutatetype	O modo de escrita de dados.	String	Não	INSERT_OR_UPDATE	Se uma chave primária estiver definida para a tabela física do Hologres, o sink do Hologres garante semântica exactly-once com base nessa chave. Caso cheguem dados com chave primária duplicada, o parâmetro `mutatetype` determina como a tabela sink será atualizada. O parâmetro `mutatetype` aceita os seguintes valores: INSERT_OR_IGNORE: Mantém a primeira ocorrência dos dados e ignora todos os subsequentes. INSERT_OR_REPLACE: Novos dados substituem toda a linha existente. INSERT_OR_UPDATE: Atualiza um subconjunto de colunas de uma linha existente. Por exemplo, se uma tabela tiver quatro colunas (a, b, c e d) e "a" for a Chave Primária (PK), ao gravar dados apenas para as colunas "a" e "b" no Hologres, caso já exista uma linha com a mesma PK, o sistema atualizará somente a coluna "b". As colunas "c" e "d" permanecerão inalteradas.
createparttable	Define se partições inexistentes devem ser criadas automaticamente ao gravar em uma tabela particionada.	Boolean	Não	false	N/A.
sink.delete-strategy	Especifica como processar mensagens de retração.	String	Não	Nenhum	Valores válidos: IGNORE_DELETE: Ignora mensagens de Update Before e Delete. Adequado para cenários onde você precisa apenas inserir ou atualizar dados, mas não excluí-los. DELETE_ROW_ON_PK: O framework Flink aplica a operação de exclusão com base na chave primária. Para operações de atualização, ele primeiro exclui os dados antigos e depois insere os novos para garantir a precisão dos dados.
jdbcWriteBatchSize	No modo JDBC, defina o número máximo de registros a serem armazenados em buffer no sink do Hologres antes de uma escrita em lote.	Integer	Não	256	Unidade: linhas. Nota Os parâmetros jdbcWriteBatchSize, jdbcWriteBatchByteSize e jdbcWriteFlushInterval possuem uma relação OU. Se você definir os três parâmetros, os dados resultantes serão gravados assim que qualquer uma das condições for atendida.
jdbcWriteBatchByteSize	No modo JDBC, este parâmetro especifica o tamanho máximo de dados em bytes que o sink do Hologres armazena em buffer antes de gravar o lote no destino.	Long	Não	2.097.152 bytes (2 MB)	Nota Os parâmetros jdbcWriteBatchSize, jdbcWriteBatchByteSize e jdbcWriteFlushInterval possuem uma relação OU. Se você definir os três parâmetros, os dados resultantes serão gravados assim que qualquer uma das condições for atendida.
jdbcWriteFlushInterval	No modo JDBC, este parâmetro especifica o tempo máximo que o sink do Hologres aguarda antes de gravar os dados em buffer no Hologres.	Long	Não	10000	Unidade: milissegundos. Nota Os parâmetros jdbcWriteBatchSize, jdbcWriteBatchByteSize e jdbcWriteFlushInterval possuem uma relação OU. Se você definir os três parâmetros, os dados resultantes serão gravados assim que a condição de qualquer um deles for atendida.
ignoreNullWhenUpdate	Define se valores nulos nos dados recebidos devem ser ignorados quando `mutatetype` estiver definido como `INSERT_OR_UPDATE`.	Boolean	Não	false	Valores válidos: false (padrão): Grava valores nulos na tabela sink do Hologres. true: Ignora valores nulos nos dados recebidos.
jdbcEnableDefaultForNotNullColumn	Define se um valor padrão deve ser inserido quando um valor nulo for gravado em uma coluna NOT NULL que não possua um valor padrão definido.	Boolean	Não	true	Valores válidos: true (padrão): Permite que o conector insira um valor padrão. As regras são as seguintes: Para colunas do tipo String, uma string vazia ("") é gravada. Para colunas do tipo Number, 0 é gravado. Para colunas dos tipos Date, timestamp ou timestamptz, 1970-01-01 00:00:00 é gravado. false: Não insere um valor padrão. Uma exceção é lançada quando um valor nulo é gravado em uma coluna NOT NULL.
remove-u0000-in-text.enabled	Define se o caractere nulo (`\u0000`) deve ser removido das strings antes da gravação.	Boolean	Não	false	Valores válidos: false (padrão): O conector não processa os dados. No entanto, se dados inconsistentes forem encontrados, uma operação de escrita poderá lançar a seguinte exceção: `ERROR: invalid byte sequence for encoding "UTF8": 0x00` Nesse caso, você deve processar os dados inconsistentes na tabela de origem ou definir uma lógica de tratamento de dados na sua instrução SQL. true: O conector remove o caractere `\u0000` das strings para evitar exceções de escrita.
deduplication.enabled	Define se a deduplicação deve ser realizada dentro de cada lote antes da gravação nos modos jdbc e jdbc_fixed.	Boolean	Não	true	Valores válidos: true (padrão): Ative a deduplicação dentro de um lote. Se vários registros tiverem a mesma chave primária, apenas o último será mantido. Por exemplo, considere dados com dois campos, onde o primeiro campo é a chave primária: Os registros `INSERT (1,'a')` e `INSERT (1,'b')` chegam sequencialmente. Após a deduplicação, apenas o registro posterior `(1,'b')` é gravado na tabela sink do Hologres. A tabela sink do Hologres já contém o registro `(1,'a')`. Se os registros `DELETE (1,'a')` e `INSERT (1,'b')` chegarem sequencialmente, apenas o último registro recebido `(1,'b')` será gravado no Hologres. Isso é tratado como uma atualização direta, e não como uma exclusão seguida de uma inserção. false: Nenhuma deduplicação é realizada durante o agrupamento em lotes. Se um novo registro tiver a mesma chave primária que um registro já presente no lote atual, o lote atual será gravado primeiro. Após a conclusão da gravação, o novo registro será processado.
sink.type-normalize-strategy	A estratégia de mapeamento de tipos de dados.	String	Não	STANDARD	Estratégia usada pelo sink do Hologres para converter tipos de dados upstream em tipos do Hologres. STANDARD: Converte tipos do Flink CDC para tipos padrão do PostgreSQL (PG). BROADEN: Converte tipos do Flink CDC para tipos do Hologres com faixa mais ampla. ONLY_BIGINT_OR_TEXT: Converte todos os tipos do Flink CDC para BIGINT ou TEXT no Hologres.
sink.insert.legacy-put-handler	Define se o Put Handler legado deve ser usado para gravar dados no Hologres.	Boolean	Não	false	Valores válidos: `false` (padrão): O novo Put Handler é usado para gravar dados. O formato SQL para a operação de escrita é `insert into xxx(c0,c1,...) select unnest(?),unnest(?),... on conflict`. true: Grava dados usando o Put Handler legado. O formato SQL para escrita é `insert into xxx(c0,c1,...) values (?,?,...),... on conflict;` .
table_property.*	As propriedades da tabela física para o Hologres.	String	Não	Nenhum	Ao criar uma tabela no Hologres, você pode definir propriedades de tabela física na cláusula WITH. Propriedades adequadas ajudam o sistema a organizar e consultar dados de forma eficiente. Aviso O parâmetro `table_property.distribution_key` tem como padrão o valor da chave primária. Não altere essa configuração a menos que compreenda totalmente o impacto, pois isso pode afetar a correção das gravações de dados.
connection.ssl.mode	Define se a criptografia de transporte Secure Sockets Layer (SSL) deve ser ativada e qual modo usar.	String	Não	disable	`disable` (padrão): Desativa a criptografia de transporte. `require`: Ative o SSL para criptografar o link de dados. `verify-ca`: Ative o SSL, criptografa o link de dados e usa um certificado CA para verificar a autenticidade do servidor Hologres. `verify-full`: Ative o SSL, criptografa o link de dados, usa um certificado CA para verificar a autenticidade do servidor Hologres e verifique se o nome CN ou DNS no certificado corresponde ao endpoint configurado do Hologres. Nota O Hologres V2.1 e versões posteriores suportam os modos verify-ca e verify-full. Para mais informações, consulte Transport encryption. Se você definir este parâmetro como `verify-ca` ou `verify-full`, também deverá configure o parâmetro `connection.ssl.root-cert.location`.
connection.ssl.root-cert.location	Quando o modo de criptografia de transporte exige um certificado, especifica o caminho para o arquivo de certificado.	String	Não	Nenhum	Quando connection.ssl.mode estiver definido como verify-ca ou verify-full, você também deve configure o caminho para o certificado CA. Você pode carregar o certificado na plataforma usando o recurso File Management no console do Realtime Compute. Após o carregamento, o certificado é armazenado no diretório /flink/usrlib. Por exemplo, se o arquivo de certificado CA for denominado certificate.crt, o valor do parâmetro deve ser `'/flink/usrlib/certificate.crt'`. Nota Para obter um certificado CA, consulte Transport encryption - Download a CA certificate.
connection.akv4.enabled	Define se o modo AKV4 deve ser ativado para conectar ao servidor Hologres.	Boolean	Não	false	N/A.
connection.akv4.region	Quando o modo AKV4 está ativado, especifica a região onde o servidor está localizado.	String	Não	Nenhum	Por exemplo, `cn-shanghai`.

Reutilizar um Catalog existente

A partir do VVR 11.5, você pode referenciar diretamente um Catalog interno do Hologres criado na página Data Management em um job de ingestão de dados Flink CDC. Isso reduz a necessidade de especificar manualmente as propriedades de conexão.

sink:
  type: hologres
  using.built-in-catalog: my_holo_catalog

Jobs de ingestão de dados podem reutilizar automaticamente os seguintes parâmetros do Catalog do Hologres:

endpoint
username
password
dbname

Para substituir esses parâmetros reutilizados automaticamente, especifique explicitamente os parâmetros YAML correspondentes, que têm prioridade mais alta.

Mapeamento de tipos de dados

Use o parâmetro sink.type-normalize-strategy para definir a estratégia de conversão de dados upstream para tipos do Hologres.

Nota

Ative sink.type-normalize-strategy ao iniciar um job YAML pela primeira vez. Se você ativá-lo após o início do job, será necessário exclua a tabela downstream e reiniciar o job sem estado para que a configuração tenha efeito.
Atualmente, tipos de array suportam apenas INTEGER, BIGINT, FLOAT, DOUBLE, BOOLEAN, CHAR e VARCHAR.
O Hologres não suporta o tipo numeric como chave primária. Se o tipo de uma chave primária for mapeado para numeric, o sistema o converterá para o tipo varchar.

STANDARD

Quando sink.type-normalize-strategy está definido como STANDARD, os tipos são mapeados da seguinte forma:

Tipo Flink CDC	Tipo Hologres
CHAR	bpchar
STRING	text
VARCHAR	text (se length > 10485760)
VARCHAR	varchar (se length <= 10485760)
BOOLEAN	bool
BINARY	bytea
VARBINARY	bytea
DECIMAL	numeric
TINYINT	int2
SMALLINT	int2
INTEGER	int4
BIGINT	int8
FLOAT	float4
DOUBLE	float8
DATE	date
TIME_WITHOUT_TIME_ZONE	time
TIMESTAMP_WITHOUT_TIME_ZONE	timestamp
TIMESTAMP_WITH_LOCAL_TIME_ZONE	timestamptz
ARRAY	Arrays dos tipos de elemento correspondentes
MAP	Não suportado
ROW	Não suportado

BROADEN

Quando sink.type-normalize-strategy está definido como BROADEN, os tipos do Flink CDC são convertidos para tipos do Hologres com faixa mais ampla. Os tipos são mapeados da seguinte forma:

Tipo Flink CDC	Tipo Hologres
CHAR	text
STRING
VARCHAR
VARCHAR
BOOLEAN	bool
BINARY	bytea
VARBINARY	bytea
DECIMAL	numeric
TINYINT	int8
SMALLINT
INTEGER
BIGINT
FLOAT	float8
DOUBLE	float8
DATE	date
TIME_WITHOUT_TIME_ZONE	time
TIMESTAMP_WITHOUT_TIME_ZONE	timestamp
TIMESTAMP_WITH_LOCAL_TIME_ZONE	timestamptz
ARRAY	Arrays dos tipos de elemento correspondentes
MAP	Não suportado
ROW	Não suportado

ONLY_BIGINT_OR_TEXT

Quando sink.type-normalize-strategy está definido como ONLY_BIGINT_OR_TEXT, todos os tipos do Flink CDC são convertidos para tipos BIGINT ou STRING no Hologres. O mapeamento de tipos de dados é o seguinte:

Tipo Flink CDC	Tipo Hologres
TINYINT	int8
SMALLINT
INTEGER
BIGINT
BOOLEAN	text
BINARY
VARBINARY
DECIMAL
FLOAT
DOUBLE
DATE
TIME_WITHOUT_TIME_ZONE
TIMESTAMP_WITHOUT_TIME_ZONE
TIMESTAMP_WITH_LOCAL_TIME_ZONE
ARRAY	Arrays dos tipos de elemento correspondentes
MAP	Não suportado
ROW	Não suportado

Gravação em tabelas particionadas

Utilize o sink do Hologres com uma transformação para gravar dados upstream em uma tabela particionada do Hologres.

A partition key deve fazer parte da primary key. Se você usar uma coluna que não seja chave primária dos dados upstream como partition key, as chaves primárias das tabelas upstream e downstream poderão ficar inconsistentes, causando discrepâncias de dados durante a sincronização.
O Hologres suporta colunas dos tipos de dados TEXT, VARCHAR e INT como partition key. A partir da versão 1.3.22, colunas do tipo de dados DATE também são suportadas.
Para crie automaticamente tabelas particionadas filhas, defina o parâmetro createparttable como true. Caso contrário, você deverá criá-las manualmente.

Consulte Writing data to a partitioned table para ver um exemplo.

Sincronização de esquema de tabela

Um pipeline CDC YAML usa diferentes estratégias para lidar com alterações no esquema da tabela, configuráveis por meio do parâmetro de nível de pipeline schema.change.behavior. Os valores válidos para schema.change.behavior são IGNORE, LENIENT, TRY_EVOLVE, EVOLVE e EXCEPTION. O Sink do Hologres atualmente não suporta a estratégia TRY_EVOLVE. As estratégias LENIENT e EVOLVE envolvem alterações no esquema da tabela. As seções a seguir descrevem como esses dois modos lidam com diferentes eventos de alteração de esquema.

LENIENT (padrão)

No modo LENIENT, as alterações de esquema são tratadas da seguinte forma:

Adicionar uma coluna anulável: A coluna correspondente é adicionada automaticamente ao final da tabela sink e seus dados são sincronizados.
Excluir uma coluna anulável: A coluna não é excluída da tabela sink. Em vez disso, ela é preenchida automaticamente com valores NULL.
Adicionar uma coluna não anulável: Uma coluna anulável correspondente é adicionada automaticamente ao final da tabela sink e seus dados são sincronizados. Para linhas existentes, essa nova coluna é preenchida automaticamente com NULL.
Renomear uma coluna: Esta operação é tratada como a exclusão de uma coluna e a adição de uma nova. Uma nova coluna com o nome especificado é adicionada ao final da tabela sink, e a coluna com o nome original é preenchida automaticamente com valores NULL. Por exemplo, se col_a for renomeada para col_b, uma coluna col_b será adicionada ao final da tabela sink, e a coluna col_a será preenchida automaticamente com valores NULL.
Alterar o tipo de dados de uma coluna: Não suportado. Como o Hologres não suporta a alteração do tipo de dados de uma coluna, você deve usar o parâmetro sink.type-normalize-strategy.
As seguintes alterações de esquema não são suportadas:
- Alterações em restrições, como chave primária ou índice.
- Exclusão de uma coluna não anulável.
- Alteração de uma coluna de NOT NULL para NULLABLE.

EVOLVE

No modo EVOLVE, as alterações de esquema são tratadas da seguinte forma:

Adicionar uma coluna anulável: Suportado.
Excluir uma coluna anulável: Não suportado.
Adicionar uma coluna não anulável: Uma nova coluna anulável é adicionada à tabela sink.
Renomear uma coluna: Suportado. A coluna original é renomeada na tabela sink.
Alterar o tipo de dados de uma coluna: Não suportado. Como o Hologres não suporta a alteração do tipo de dados de uma coluna, você deve usar o parâmetro sink.type-normalize-strategy.
As seguintes alterações de esquema não são suportadas:
- Alterações em restrições, como chave primária ou índice.
- Exclusão de uma coluna não anulável.
- Alteração de uma coluna de NOT NULL para NULLABLE.

Aviso

No modo EVOLVE, realizar uma reinicialização sem estado sem exclua a tabela sink pode causar falha no pipeline devido a inconsistências de esquema entre os dados upstream e a tabela sink. Nesse caso, você deve ajustar manualmente o esquema da tabela sink.

Consulte Enabling the EVOLVE mode para ver um exemplo.

Exemplos de código

Ampliação de tipos

Use o parâmetro sink.type-normalize-strategy para configure a ampliação de tipos.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Gravação em uma tabela particionada

Converta o campo de timestamp create_time para um tipo de data e use-o como chave de partição para a tabela do Hologres.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Automatically create partitioned tables if they do not exist.
  createparttable: true
 
transform:
  - source-table: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key 

pipeline:
  name: MySQL to Hologres Pipeline

Ativando o modo EVOLVE****

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Automatically create partitioned tables if they do not exist.
  createparttable: true

pipeline:
  name: MySQL to Hologres Pipeline
  schema.change.behavior: evolve

Sincronização de tabela única

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Sincronização de banco de dados completo

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Mesclagem de tabelas fragmentadas

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.  
  sink.type-normalize-strategy: BROADEN
  
route:
  # All sharded tables in the MySQL test_db database are merged into a single Hologres table named test_db.user.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Hologres Pipeline

Sincronização para um esquema específico

No Hologres, um esquema corresponde a um banco de dados no MySQL. Você pode especifique o esquema para as tabelas sink.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN
  
route:
  # Synchronize all tables from the MySQL test_db database to the Hologres test_db2 schema while keeping the original table names.
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Hologres Pipeline

Sincronização de novas tabelas sem reinicialização

Para sincronizar tabelas recém-adicionadas em tempo real enquanto um job está em execução, defina scan.binlog.newly-added-table.enable = true.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  # Automatically captures new tables created while the job is running.  
  scan.binlog.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Adição de tabelas existentes na reinicialização

Para incluir uma tabela existente na sincronização, defina scan.newly-added-table.enabled como true e execute o job.

Aviso

Não use scan.newly-added-table.enabled em um job que foi executado anteriormente com scan.binlog.newly-added-table.enabled. Essa combinação causa duplicação de dados na reinicialização.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # On restart, the job scans for new tables matching the `tables` parameter and performs a snapshot.
  # Note: This parameter must be used with scan.startup.mode: initial.
  scan.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Exclusão de tabelas

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Excludes tables that match this regular expression.
  tables.exclude: test_db.table1
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Documentos relacionados

Para saber mais sobre os módulos source, sink, transform e route, consulte Flink CDC Data Ingestion Job Development Reference.
Para desenvolver um job YAML de ingestão de dados, consulte Flink CDC Data Ingestion Job Development.