Como usar o flink-dts-connector - Data Transmission Service: Central de Documentação do Alibaba Cloud

Observações de uso

Este conector oferece suporte a clientes Flink que usam a DataStream API, Table API ou SQL.
Se o seu cliente Flink usar a Table API ou SQL, você poderá consumir dados de apenas uma única tabela por configuração. Para consumir dados de várias tabelas, é necessário configurar e executar várias tarefas independentes.

Procedimento

Este tópico usa o IntelliJ IDEA Community Edition 2020.1 para Windows para demonstrar como usar o flink-dts-connector para consumir dados de uma instância de assinatura.

crie uma tarefa de rastreamento de alterações. Para mais informações, consulte os tópicos relevantes em Change tracking overview.
crie um ou mais grupos de consumidores. Para mais informações, consulte Add a consumer group.
baixe e descompacte o flink-dts-connector.
Abra o IntelliJ IDEA e clique em Open or Import.
Na caixa de diálogo exibida, acesse o diretório descompactado do flink-dts-connector, expanda as pastas e localize o arquivo Project Object Model (pom.xml).
Na caixa de diálogo exibida, selecione Open as Project.

Adicione a seguinte dependência ao arquivo pom.xml:

<dependency>
      <groupId>com.alibaba.flink</groupId>
      <artifactId>flink-dts-connector</artifactId>
      <version>1.1.1-SNAPSHOT</version>
      <classifier>jar-with-dependencies</classifier>
</dependency>

No IntelliJ IDEA, expanda as pastas e selecione o arquivo Java correspondente ao tipo de programa do seu conector Flink.
- Se o seu cliente Flink usar a DataStream API, clique duas vezes no arquivo flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\datastream\DtsExample.java e execute as seguintes etapas:
  1. Na parte superior da janela do IntelliJ IDEA, clique em ícone execute.
  2. No menu exibido, clique em DtsExample > Edit.
  3. No campo Program arguments, insira os parâmetros e seus valores com base no exemplo a seguir. Em seguida, clique em execute para iniciar o flink-dts-connector.
    
    Nota
    Para mais informações sobre os parâmetros e como obter seus valores, consulte Parameters.
```
--broker-url dts-cn-******.******.***:****** --topic cn_hangzhou_rm_**********_dtstest_version2 --sid dts****** --user dtstest --password Test123456 --checkpoint 1624440043
```
  4. A saída confirma que o cliente pode assinar as alterações de dados do banco de dados source.
    
    Nota
    Para consultar registros específicos de alteração de dados, faça login na interface do Task Manager do seu cliente Flink.
- Se o seu cliente Flink usar a Table API ou SQL, clique duas vezes no arquivo flink-dts-connector-master\src\test\java\com\alibaba\flink\connectors\dts\sql\DtsTableISelectTCaseTest.java e execute as seguintes etapas:
  
  Nota
  Cada arquivo DtsTableISelectTCaseTest.java pode ser configurado para consumir dados de apenas uma tabela. Para configurar e consumir dados de várias tabelas, você precisa repetir a configuração e executar várias tarefas independentes.
  1. Adicione duas barras (//) para comentar a linha de código especificada.
  2. configure as informações da única tabela da qual você deseja consumir dados. Há suporte para instruções SQL.
  3. configure os parâmetros da instância de assinatura. Para mais informações sobre os parâmetros e como obter seus valores, consulte Parameters.
  4. Na parte superior da janela do IntelliJ IDEA, clique em execute'DtsTableISelectTCaseTest' para iniciar o flink-dts-connector.
  5. A saída confirma que o cliente pode assinar as alterações de dados do banco de dados source.
    
    Nota
    Para consultar registros específicos de alteração de dados, faça login na interface do Task Manager do seu cliente Flink.

Parâmetros

DtsExample.java	DtsTableISelectTCaseTest.java	Descrição	Método de consulta
`broker-url`	`dts.server`	O endpoint de rede e a porta da instância de assinatura. Nota Se a instância ECS onde seu cliente Flink está implantado e a instância de assinatura estiverem na mesma rede clássica ou Virtual Private cloud (VPC), recomendamos o uso de um endpoint interno para assinatura de dados, a fim de minimizar a latência de rede. Não recomendamos o uso de um endpoint público devido à possível instabilidade de rede.	No console do DTS, clique em ID da instância de assinatura de destino. Na página Basic Information, você pode obter as informações de Topic e Network.
`topic`	`topic`	O tópico da instância de assinatura.
`sid`	`dts.sid`	O ID do grupo de consumidores.	No console do DTS, clique em ID da instância de assinatura de destino. No painel de navegação à esquerda, clique em Consume Data. Você pode obter o Consumer Group ID/Name e a Account do grupo de consumidores. Nota A senha da conta do grupo de consumidores é especificada quando você cria o grupo de consumidores.
`user`	`dts.user`	A conta do grupo de consumidores. Aviso Se você não estiver usando o flink-dts-connector fornecido neste tópico, defina o nome de usuário no formato `<consumer_group_account>-<consumer_group_ID>`, como `dtstest-dtsaebpv`. Caso contrário, a conexão falhará.
`password`	`dts.password`	A senha da conta.
`checkpoint`	`dts.checkpoint`	O offset do consumidor. Este parâmetro especifica o timestamp Unix do primeiro registro de dados a ser consumido. Exemplo: 1624440043. Nota Você pode usar o offset do consumidor para: Retomar o consumo a partir do último offset após uma interrupção do aplicativo, evitando a perda de dados. especifique um offset para começar a consumir dados a partir de um ponto específico no tempo.	O horário de início do consumo deve estar dentro do intervalo de dados da instância de assinatura e deve ser convertido em um timestamp Unix. Nota Você pode visualizar o intervalo de dados da instância de assinatura na coluna Data Range da lista de tarefas de assinatura. Você pode usar um mecanismo de busca para encontrar uma ferramenta de conversão de timestamp Unix.
N/D	`dts-cdc.table.name`	O objeto assinado. Você pode especificar apenas uma única tabela. Os requisitos de formato são os seguintes: Se o Database Type for MySQL, PolarDB for MySQL, PolarDB-X 1.0 ou PolarDB-X 2.0, use o formato `<database_name>.<table_name>`. Para outros tipos de banco de dados, use o formato `<schema_name>.<table_name>`.	No console do DTS, clique em ID da instância de assinatura de destino. Na parte superior da página Basic Information ou Task Management, clique em View Objects para encontrar o banco de dados e a tabela assinados.

Perguntas frequentes

Mensagem de erro	Possível causa	Solução
`Cluster changed from * to *, consumer require restart.`	Um failover no módulo DStore, que o Data Transmission Service (DTS) usa para ler dados incrementais, faz com que o offset do consumidor do cliente Flink seja perdido.	Não reinicie o cliente. Em vez disso, consulte o offset do consumidor do cliente e passe o valor novamente para o parâmetro `checkpoint` ou `dts.checkpoint` no arquivo DtsExample.java ou DtsTableISelectTCaseTest.java para retomar o consumo de dados.