AnalyticDB:Sincronizar dados do Kafka com o recurso de sincronização de dados do APS (Recomendado)

Limitações

• O sistema aceita apenas mensagens do Kafka em formato json. Outros formatos causam erros de sincronização.

• Alterações no schema da tabela do Kafka não se propagam automaticamente para o AnalyticDB for MySQL. Aplique as mudanças de DDL manualmente.

• A API do Kafka trunca dados de amostra maiores que 8 KB, o que impede a análise da amostra e a geração automática de mapeamentos de campos.

• Os dados gravados no AnalyticDB for MySQL só ficam visíveis após a execução de uma operação de Commit. Como o intervalo padrão de Commit é de 5 minutos, aguarde pelo menos esse período após iniciar a tarefa antes de consultar o primeiro lote de dados.

• Caminhos do OSS em tarefas de sincronização diferentes não podem compartilhar um prefixo. Por exemplo, oss://testBucketName/test/sls1/ e oss://testBucketName/test/ entram em conflito e causam a sobrescrita de dados.

• Se uma tarefa de sincronização falhar e os dados do tópico do Kafka tiverem expirado, não será possível recuperar as informações removidas ao reiniciar a tarefa. Para reduzir esse risco, aumente o período de retenção de dados do tópico. Em caso de falha na tarefa, entre em contato com o suporte técnico imediatamente.

Pré-requisitos

Antes de começar, verifique se você possui:

• Um cluster do AnalyticDB for MySQL nas edições Enterprise, Basic ou Data Lakehouse

• Um grupo de recursos de job

• Uma conta de banco de dados para o cluster (consulte a tabela abaixo)

• Uma instância do ApsaraMQ for Kafka na mesma região do cluster do AnalyticDB for MySQL

• Um tópico do Kafka com mensagens já enviadas (consulte Início rápido do ApsaraMQ for Kafka)

Requisitos de conta de banco de dados por tipo de conta:

Faturamento

O uso do recurso de sincronização de dados gera as seguintes taxas:

• AnalyticDB for MySQL: Taxas de recursos ACU elásticos. Consulte Faturamento da Edição Data Lakehouse e Faturamento das Edições Enterprise e Basic.

• Object Storage Service (OSS): Taxas de armazenamento, taxas de requisição GET e taxas de requisição PUT e outras. Consulte Visão geral do faturamento.

Como funciona

1. Adicione uma source de dados Kafka para identificar a instância e o tópico do ApsaraMQ for Kafka de onde os dados serão lidos.

2. Crie um link de dados (job de sincronização) que mapeie as mensagens do Kafka para uma tabela do AnalyticDB for MySQL e configure a análise de json, as chaves de partição e o offset do consumidor.

3. Inicie a tarefa. Ela começa a consumir dados do Kafka a partir do offset selecionado e os grava na tabela de destino.

4. Consulte os dados ingeridos usando Spark SQL.

Etapa 1: Criar uma source de dados

Ignore esta etapa se você já adicionou uma source de dados Kafka . Vá diretamente para a Etapa 2: Criar um link de dados .

1. Faça login no console do AnalyticDB for MySQL. No canto superior esquerdo, selecione uma região. No painel de navegação à esquerda, clique em Clusters e, em seguida, clique no ID do cluster.

2. No painel de navegação à esquerda, escolha Data Ingestion > Data Sources.

3. Clique em Create Data Source.

4. Na página Create Data Source, configure os seguintes parâmetros:

   Parâmetro	Descrição
   Data Source Type	Selecione Kafka.
   Data Source Name	Gerado automaticamente com base no tipo de source e na hora atual. Altere conforme necessário.
   Data Source Description	Insira uma descrição, como o cenário de negócios ou o escopo dos dados.
   Deployment Mode	Apenas Alibaba Cloud Instance é compatível.
   Kafka Instance	O ID da instância Kafka. Encontre-o na página Instances do console do ApsaraMQ for Kafka.
   Kafka Topic	O nome do tópico. Localize-o na página Topics da instância de destino no console do ApsaraMQ for Kafka.
   Message Data Format	Apenas json é compatível.

5. Clique em Create.

Etapa 2: Criar um link de dados

1. No painel de navegação à esquerda, clique em Simple Log Service/Kafka Data Synchronization.

2. Clique em Create Synchronization Job.

3. Na página Create Synchronization Job, configure as três seções abaixo.

Etapa 3: Iniciar a tarefa de sincronização de dados

1. Na página Simple Log Service/Kafka Data Synchronization, localize a tarefa criada e clique em Start na coluna Actions.

2. Clique em Search. A tarefa foi iniciada com sucesso quando seu status mudar para Running.

O primeiro lote de dados estará disponível para consulta pelo menos 5 minutos após o início da tarefa, pois os dados são confirmados em intervalos de 5 minutos por padrão.

Etapa 4: Analisar os dados

Após a sincronização dos dados, use Spark SQL para consultá-los no AnalyticDB for MySQL. Para mais informações, consulte Editor de desenvolvimento Spark e Desenvolvimento de aplicações Spark offline.

1. No painel de navegação à esquerda, escolha Job Development > Spark JAR Development.

2. Insira suas instruções Spark SQL no modelo padrão e clique em Run Now. Veja um exemplo abaixo:

   -- Example of Spark SQL. Modify the content and run your Spark program.
   
   conf spark.driver.resourceSpec=medium;
   conf spark.executor.instances=2;
   conf spark.executor.resourceSpec=medium;
   conf spark.app.name=Spark SQL Test;
   conf spark.adb.connectors=oss;
   
   -- SQL statements
   show tables from lakehouse20220413156_adbTest;

3. (Opcional) Na aba Applications, clique em Logs na coluna Actions para visualizar os logs de execução do Spark SQL.

Etapa 5 (Opcional): Gerenciar a source de dados

Acesse Data Ingestion > Data Sources. As seguintes operações estão disponíveis na coluna Actions:

Níveis de análise de json e inferência de schema

A configuração Parsed JSON Layers controla quantos níveis aninhados de uma mensagem json são expandidos em campos de destino separados.

Mensagem de exemplo:

{
  "name" : "zhangle",
  "age" : 18,
  "device" : {
    "os" : {
        "test":lag,
        "member":{
             "fa":zhangsan,
             "mo":limei
           }
         },
    "brand" : "none",
    "version" : "11.4.2"
  }
}

Nível 0 — Sem análise. A mensagem json inteira é enviada como um único campo.

Nível 1 (padrão) — Os campos de nível superior são expandidos.

Nível 2 — Dois níveis expandidos. Campos não aninhados são enviados diretamente; campos aninhados se expandem para seus subcampos.

Nível 3

Nível 4

Próximos passos

• Editor de desenvolvimento Spark

• Desenvolvimento de aplicações Spark offline

• Faturamento da Edição Data Lakehouse

• Faturamento das Edições Enterprise e Basic