MaxCompute:Criar uma tabela externa Apache Paimon usando o Realtime Compute for Apache Flink

Informações básicas

O Apache Paimon é um formato unificado de armazenamento em data lake para processamento em streaming e em lote que oferece gravações de alto throughput e consultas de baixa latência. Mecanismos de computação comuns, como Spark, Hive e Trino no Realtime Compute for Apache Flink e no E-MapReduce, integram-se perfeitamente ao Paimon. Com o Apache Paimon, você constrói rapidamente um data lake no Object Storage Service (OSS) e o conecta ao MaxCompute para análises. Para mais informações sobre o Apache Paimon, consulte a documentação do Apache Paimon.

Pré-requisitos

• A conta Alibaba Cloud usada nas operações deve ter a permissão CreateTable para criar tabelas no MaxCompute. Para mais informações sobre permissões de tabela, consulte Permissões do MaxCompute.

• Crie um projeto do MaxCompute. Para mais informações, consulte Criar um projeto.

• Ative o OSS e crie um bucket e um diretório de arquivos. Para mais informações, consulte Criar um bucket.

  Nota

  O MaxCompute está disponível apenas em regiões específicas. Para evitar problemas de conectividade de dados entre regiões, use um bucket na mesma região do seu projeto do MaxCompute.

• Ative o Flink totalmente gerenciado. Para mais informações, consulte Criar um workspace.

Etapa 1: Carregar o plug-in Apache Paimon no projeto do MaxCompute

Use um dos métodos a seguir para carregar o Plug-in do Paimon no projeto do MaxCompute.

ADD JAR <path_to_paimon_maxcompute_connector.jar>;

Etapa 2: Criar uma tabela externa Apache Paimon usando o Realtime Compute for Apache Flink

Esta melhor prática baseia-se no Realtime Compute for Apache Flink. O serviço grava dados de arquivos Apache Paimon no OSS. Crie um catálogo Apache Paimon no console do Realtime Compute for Apache Flink e, nesse catálogo, crie uma tabela Apache Paimon que o MaxCompute possa usar para ler os dados no OSS. Em seguida, o MaxCompute usa essa tabela Apache Paimon como tabela externa para ler os dados armazenados no OSS.

1. Faça login no console do Realtime Compute for Apache Flink e crie um Script. Para mais informações, consulte Scripts de consulta.

2. No editor de scripts, insira a definição do catálogo e os valores dos parâmetros a seguir. Selecione o código e clique em Run.

   CREATE CATALOG `<catalog name>` WITH (
    'type' = 'paimon',
     'metastore' = 'maxcompute',
     'warehouse' = '<warehouse>',
     'maxcompute.endpoint' = '<maxcompute.endpoint>',
     'maxcompute.project' = '<maxcompute.project>',
     'maxcompute.accessid' = '<maxcompute.accessid>',
     'maxcompute.accesskey' = '<maxcompute.accesskey>',
     'maxcompute.oss.endpoint' = '<maxcompute.oss.endpoint>',
     'fs.oss.endpoint' = '<fs.oss.endpoint>',
     'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
     'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
   );

   A tabela a seguir descreve os parâmetros do código.

   Parâmetro	Obrigatório	Descrição
   catalog name	Sim	Nome do catálogo Apache Paimon. Deve conter apenas letras. Neste tópico, o nome do catálogo é catalogname.
   type	Sim	Tipo do catálogo. Defina o valor como paimon.
   metastore	Sim	Tipo de armazenamento de metadados. Defina o valor como maxcompute.
   warehouse	Sim	Diretório do data warehouse no OSS, no formato oss://<bucket>/<object>. bucket: nome do bucket OSS criado. object: caminho onde os dados estão armazenados. Visualize os nomes do bucket e do objeto no console do OSS.
   maxcompute.endpoint	Sim	Endpoint do serviço MaxCompute. Configure este parâmetro com base na região e no tipo de conexão de rede selecionados durante a criação do projeto do MaxCompute. Para mais informações sobre os endpoints correspondentes a diferentes regiões e tipos de rede, consulte Endpoint.
   maxcompute.project	Sim	Nome do projeto do MaxCompute. Projetos do MaxCompute com o recurso de schema habilitado não são suportados.
   maxcompute.accessid	Sim	AccessKey ID da conta Alibaba Cloud ou do usuário RAM com permissões no MaxCompute. Obtenha o AccessKey ID na página Gerenciamento de AccessKey.
   maxcompute.accesskey	Sim	AccessKey secret correspondente ao AccessKey ID.
   maxcompute.oss.endpoint	Não	Endpoint do OSS acessado pelo MaxCompute. Se você não configurar este parâmetro, o sistema usará o valor do parâmetro fs.oss.endpoint por padrão. Importante O bucket OSS deve residir na mesma região do projeto do MaxCompute. Recomendamos definir o parâmetro maxcompute.oss.endpoint com um endpoint interno. Para mais informações sobre os endpoints do OSS para diferentes tipos de rede em cada região, consulte Regiões e Endpoints.
   fs.oss.endpoint	Não	Endpoint do OSS. Este parâmetro é obrigatório se o bucket OSS especificado no parâmetro warehouse não estiver na mesma região do workspace do Realtime Compute for Apache Flink ou se você usar um bucket OSS de outra conta Alibaba Cloud. Nota Configure o endpoint com base na região e no método de conexão de rede selecionados durante a criação do bucket OSS. Para mais informações sobre os endpoints correspondentes a diferentes regiões e tipos de rede, consulte Regiões e Endpoints.
   fs.oss.accessKeyId	Não	AccessKey ID da conta Alibaba Cloud ou do usuário RAM com permissões de leitura e gravação no OSS. Este parâmetro é obrigatório se o bucket OSS especificado no parâmetro warehouse não estiver na mesma região do workspace do Realtime Compute for Apache Flink ou se você usar um bucket OSS de outra conta Alibaba Cloud. Obtenha o AccessKey ID na página Gerenciamento de AccessKey.
   fs.oss.accessKeySecret	Não	AccessKey secret correspondente ao AccessKey ID. Este parâmetro é obrigatório se o bucket OSS especificado no parâmetro warehouse não estiver na mesma região do workspace do Realtime Compute for Apache Flink ou se você usar um bucket OSS de outra conta Alibaba Cloud.

3. Crie uma tabela Apache Paimon.

   1. Crie uma tabela chamada test_tbl.

      Na aba Script, execute a instrução a seguir no editor de scripts. Aguarde até que uma mensagem de conclusão apareça na aba Result. Este exemplo utiliza uma tabela chamada test_tbl.

      CREATE TABLE `catalogname`.`default`.test_tbl (
       dt STRING,
       id BIGINT,
       data STRING,
       PRIMARY KEY (dt, id) NOT ENFORCED
      ) PARTITIONED BY (dt);

   2. Grave dados na tabela test_tbl.

      Na página Drafts, crie um job SQL contendo as instruções abaixo. Em seguida, faça o deploy e execute o job. Para mais informações sobre como criar e executar jobs SQL, consulte Visão geral do desenvolvimento de jobs.

      -- The checkpoint interval is set to 10s to commit data faster.
      SET 'execution.checkpointing.interval' = '10s';
      INSERT INTO `catalogname`.`default`.test_tbl VALUES ('2023-04-21', 1, 'AAA'), ('2023-04-21', 2, 'BBB'), ('2023-04-22', 1, 'CCC'), ('2023-04-22', 2, 'DDD');

      Nota

      • A tabela de resultados Apache Paimon confirma (commits) os dados sempre que um checkpoint é concluído.

      • No ambiente de produção, o intervalo de checkpoint e o intervalo mínimo entre checkpoints variam conforme os requisitos de latência do seu negócio. Geralmente, esses valores são definidos entre 1 e 10 minutos.

      • A versão do mecanismo do rascunho SQL deve ser vvr-8.0.5-flink-1.17 ou superior.

Etapa 3: Ler dados com o MaxCompute

1. Execute os comandos a seguir no Cliente local (odpscmd) ou em outra ferramenta capaz de executar instruções SQL do MaxCompute:

   SET odps.sql.common.table.planner.ext.hive.bridge = true;
   SET odps.sql.hive.compatible = true;

2. Execute o comando abaixo para consultar dados da tabela externa Apache Paimon test_tbl:

   SELECT * FROM test_tbl WHERE dt = '2023-04-21';

   O resultado retornado é:

   +------------+------------+------------+
   | id | data | dt |
   +------------+------------+------------+
   | 1 | AAA | 2023-04-21 |
   | 2 | BBB | 2023-04-21 |
   +------------+------------+------------+