Hologres:Import data from a MaxCompute partitioned table

Pré-requisitos

• Você comprou e ativou uma instância do Hologres. Para obter mais informações, consulte Purchase a Hologres instance.

• Você ativou o MaxCompute e criou um projeto. Para obter mais informações, consulte Activate MaxCompute.

• Você ativou o DataWorks e criou um workspace. Para obter mais informações, consulte Create a workspace.

Contexto

Usar uma tabela estrangeira (foreign table) no Hologres para acessar dados do MaxCompute é um padrão comum de ingestão de dados. Para importações recorrentes, configure um único job periódico que automatiza o fluxo de trabalho de importação com os recursos avançados de agendamento e orquestração de jobs do DataWorks. Para obter mais informações, consulte DataWorks job example.

Devido à complexidade do job, use o assistente de migração do DataWorks para importar o arquivo de exemplo do job para o seu projeto. Isso fornece um fluxo de trabalho pronto para uso. Em seguida, personalize-o ao alterar os parâmetros ou scripts para atender às necessidades específicas da sua empresa. Para obter mais informações, consulte Import the DataWorks job using the migration assistant.

Observações de uso

• Utiliza-se uma tabela temporária para garantir a atomicidade. O processo anexa os dados à tabela particionada apenas após concluir a importação. Isso evita a necessidade de operações de limpeza manuais, como a exclusão de tabelas, se uma tarefa de importação falhar.

• Ao atualizar dados em uma partição filha, exclua a tabela filha antiga e anexe a nova tabela temporária dentro de uma única transação para garantir a consistência transacional.

• Atenda às seguintes condições para importar um job do DataWorks com o assistente de migração:

  • Sua edição do DataWorks deve ser a Standard Edition ou superior. Para obter mais informações, consulte DataWorks editions.

  • O workspace do DataWorks deve estar vinculado às fontes de dados do MaxCompute e do Hologres. Para obter mais informações, consulte Configure a workspace.

  • O workspace do DataWorks ainda deve utilizar o DataStudio legado. O assistente de migração não suporta workspaces atualizados para o novo Data Studio. Em um workspace atualizado, a importação do pacote de recursos falha na verificação com a mensagem The migration assistant does not support workspaces that use the new Data Studio. Nesse caso, crie o job manualmente ao seguir o Procedure neste tópico.

Procedimento

1. Prepare os dados do MaxCompute

   1. Faça logon no console do DataWorks, alterne para a região de destino e, no painel de navegação à esquerda, escolha Data Analysis and Service > Data Analysis. Clique em Go to Data Analysis. No painel de navegação à esquerda, clique no ícone para acessar a página SQL Query.

   2. Na página SQL Query, insira a seguinte instrução SQL para criar uma tabela particionada e clique em Run.

      DROP TABLE IF EXISTS odps_sale_detail;
      -- Create a partitioned table named sale_detail.
      CREATE TABLE IF NOT EXISTS odps_sale_detail 
      (
          shop_name STRING
          ,customer_id STRING
          ,total_price DOUBLE
      )
      PARTITIONED BY 
      (
          sale_date STRING
      )
      ;

   3. Na página SQL Query, insira a seguinte instrução SQL para importar dados para a tabela particionada e clique em Run.

      -- Add the 20210815 partition to the source table.
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20210815')
      ;
      -- Write data to the partition.
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20210815') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3)
      ;
      -- Add the 20210816 partition to the source table.
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20210816')
      ;
      -- Write data to the partition.
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20210816') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3)
      ;
      -- Add the 20210817 partition to the source table.
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20210817')
      ;
      -- Write data to the partition.
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20210817') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3)
      ;
      -- Add the 20210818 partition to the source table.
      ALTER TABLE odps_sale_detail ADD IF NOT EXISTS PARTITION(sale_date='20210818')
      ;
      -- Write data to the partition.
      INSERT OVERWRITE TABLE odps_sale_detail PARTITION(sale_date='20210818') VALUES 
      ('s1','c1',100.1),
      ('s2','c2',100.2),
      ('s3','c3',100.3)
      ;

2. Crie tabelas no Hologres

   • Crie uma tabela estrangeira

     1. Faça logon no banco de dados.

        1. Na página de desenvolvimento do HoloWeb, clique em Metadata Management.

        2. Na página Metadata Management, clique duas vezes no nome do banco de dados que você criou na árvore de diretórios à esquerda e clique em OK.

     2. Crie uma tabela estrangeira

        1. Na página SQL Editor, clique no ícone no canto superior esquerdo para criar uma nova consulta SQL.

        2. Na nova página Ad-hoc Query, selecione o seu Instance Name e Database. No editor SQL, insira a seguinte instrução e clique em Run.

           DROP FOREIGN TABLE IF EXISTS odps_sale_detail;
           -- Create a foreign table.
           IMPORT FOREIGN SCHEMA maxcompute_project LIMIT to
           (
               odps_sale_detail
           ) 
           FROM SERVER odps_server INTO public 
           OPTIONS(if_table_exist 'error',if_unsupported_type 'error');

   • Crie uma tabela particionada (tabela interna)

     1. Faça logon no banco de dados.

        1. Na página de desenvolvimento do HoloWeb, clique em Metadata Management.

        2. Na página Metadata Management, clique duas vezes no nome do banco de dados que você criou na árvore de diretórios à esquerda e clique em OK.

     2. Crie uma tabela particionada

        1. Na página SQL Editor, clique no ícone no canto superior esquerdo para criar uma nova consulta SQL.

        2. Na nova página Ad-hoc Query, selecione o seu Instance Name e Database. No editor SQL, insira a seguinte instrução e clique em Run.

           DROP TABLE IF EXISTS holo_sale_detail;
           -- Create a Hologres partitioned table (internal table).
           BEGIN ;
           CREATE TABLE IF NOT EXISTS holo_sale_detail
           (
               shop_name TEXT
               ,customer_id TEXT 
               ,total_price FLOAT8
               ,sale_date TEXT
           )
           PARTITION BY LIST(sale_date);
           COMMIT;

3. Importe os dados da partição para uma tabela temporária do Hologres

   Na página Ad-hoc Query, insira a seguinte instrução no editor SQL e clique em Run.

   Esta instrução SQL importa a partição 20210816 da tabela particionada odps_sale_detail no projeto hologres_test do MaxCompute para a partição 20210816 da tabela particionada holo_sale_detail no Hologres.

   Nota

   O Hologres V2.1.17 e versões posteriores suportam o Serverless Computing. Para cenários como a importação de dados offline em grande escala, grandes jobs ETL e consultas de alto volume em tabelas estrangeiras, execute essas tarefas com o Serverless Computing. Esse recurso consome recursos serverless adicionais em vez dos recursos da sua instância, o que melhora a estabilidade da instância e reduz a probabilidade de erros de falta de memória (OOM). Você não precisa reservar recursos de computação extras para a sua instância e é cobrado apenas pelas tarefas que executa. Para obter mais informações sobre o Serverless Computing, consulte Serverless Computing. Para obter instruções sobre como usar o Serverless Computing, consulte Use Serverless Computing.

   -- Clean up any potential temporary tables.
   BEGIN ;
   DROP TABLE IF EXISTS holo_sale_detail_tmp_20210816;
   COMMIT ;
   -- Create a temporary table.
   SET hg_experimental_enable_create_table_like_properties=on;
   BEGIN ;
   CALL HG_CREATE_TABLE_LIKE ('holo_sale_detail_tmp_20210816', 'select * from holo_sale_detail'); 
   COMMIT;
   -- (Optional) We recommend that you use Serverless Computing to run large-scale offline data imports and ETL jobs.
   SET hg_computing_resource = 'serverless';
   -- Insert data into the temporary table.
   INSERT INTO holo_sale_detail_tmp_20210816
   SELECT *
   FROM public.odps_sale_detail
   WHERE sale_date='20210816';
   -- Reset the configuration to ensure that non-essential SQL queries do not use serverless resources.
   RESET hg_computing_resource;

4. Anexe a tabela temporária à tabela particionada do Hologres

   Na página Ad-hoc Query, insira a seguinte instrução no editor SQL e clique em Run.

   • Se houver uma tabela filha antiga, exclua-a primeiro e, em seguida, anexe a tabela temporária à tabela particionada do Hologres.

     Esta instrução SQL exclui a tabela filha holo_sale_detail_20210816 e vincula a tabela temporária holo_sale_detail_tmp_20210816 à partição 20210816 da tabela particionada holo_sale_detail.

     -- Replace the child table if it exists.
     BEGIN ;
     -- Delete the old child table.
     DROP TABLE IF EXISTS holo_sale_detail_20210816;
     -- Rename the temporary table.
     ALTER TABLE holo_sale_detail_tmp_20210816 RENAME TO holo_sale_detail_20210816;
     -- Attach the temporary table to the specified partitioned table.
     ALTER TABLE holo_sale_detail ATTACH PARTITION holo_sale_detail_20210816
     FOR VALUES IN ('20210816')
     ;
     COMMIT ;

   • Se não existir uma tabela filha antiga, anexe a tabela temporária diretamente à tabela particionada do Hologres.

     Esta instrução SQL vincula a tabela temporária holo_sale_detail_tmp_20210816 à partição 20210816 da tabela particionada holo_sale_detail.

     BEGIN ;
     -- Rename the temporary table.
     ALTER TABLE holo_sale_detail_tmp_20210816 RENAME TO holo_sale_detail_20210816;
     -- Attach the temporary table to the specified partitioned table.
     ALTER TABLE holo_sale_detail ATTACH PARTITION holo_sale_detail_20210816
     FOR VALUES IN ('20210816');
     COMMIT ;

5. Execute o ANALYZE na tabela particionada do Hologres

   Na página Ad-hoc Query, insira a seguinte instrução no editor SQL e clique em Run.

   Esta instrução SQL executa o ANALYZE na tabela particionada holo_sale_detail para coletar estatísticas, o que ajuda o planejador de consultas a gerar planos de execução eficientes. Ao executar o ANALYZE em uma tabela particionada, execute-o apenas na tabela pai.

   -- Run ANALYZE on the parent table of the partitioned table after a large data import.
   ANALYZE holo_sale_detail;

6. Limpe as partições filhas expiradas (opcional)

   Em um ambiente de produção, os dados possuem um ciclo de vida, o que exige a limpeza das partições que ultrapassaram o período de retenção.

   Na página Ad-hoc Query, insira a seguinte instrução no editor SQL e clique em Run.

   Esta instrução SQL limpa a partição 20210630.

   DROP TABLE IF EXISTS holo_sale_detail_20210630;

Inicialização completa e atualizações incrementais

A importação de dados em um clique do MaxCompute no HoloWeb fornece apenas um menu suspenso Partition Field e um único seletor Business Date, portanto, ela importa apenas uma partição por vez e não consegue importar todas as partições históricas simultaneamente. O procedimento anterior cria uma tabela estrangeira do MaxCompute no Hologres, utiliza instruções SQL para gravar os dados em uma tabela temporária (o que garante a atomicidade da importação) e, em seguida, anexa essa tabela temporária à tabela particionada (tabela interna). Como alternativa, sincronize os dados diretamente pelo Data Integration do DataWorks sem escrever SQL. Siga estas duas fases para importar todos os dados das partições históricas e manter as atualizações incrementais diárias.

Antes de começar, certifique-se de que o seu workspace do DataWorks esteja vinculado às fontes de dados do MaxCompute e do Hologres. Para obter mais informações, consulte a seção Usage notes deste tópico.

1. Inicialização completa: no console do HoloWeb, acesse o Data Integration do DataWorks pela entrada Data Migration and Synchronization. Crie uma tarefa de sincronização em lote. Selecione a tabela particionada do MaxCompute como fonte e a tabela do Hologres como destino. Defina a condição do filtro como 1=1 para importar todos os dados de partições históricas em uma única operação. Envie e execute essa tarefa de sincronização para concluir a inicialização completa. O tempo necessário depende do volume de dados das partições históricas. Depois que a importação terminar, execute o ANALYZE na tabela pai da tabela particionada para atualizar suas estatísticas de desempenho.

2. Atualizações incrementais: em um ambiente de produção, o sistema importa novas partições continuamente por meio de um agendamento recorrente. Configure o agendamento recorrente para essa tarefa de sincronização com o parâmetro bizdate para importar apenas a partição do dia atual a cada dia. Publique a tarefa antes de iniciá-la; visualize o status dela nos detalhes de execução. Para obter uma orquestração de job de agendamento mais completa (incluindo tabelas de parâmetros e limpeza de partições expiradas), consulte a seguinte seção DataWorks job example.

Exemplo de job do DataWorks

Módulos do fluxo de trabalho

• Parâmetros básicos

  Os parâmetros básicos gerenciam todos os parâmetros usados no fluxo de trabalho. Os principais parâmetros são listados a seguir.

  Id	Parâmetro	Tipo	Valor	Descrição
  1	datepre31	variável	${yyyymmdd-31}	Um parâmetro que controla a limpeza de partições expiradas. Esse valor especifica a exclusão de partições de 31 dias atrás.
  2	datetime1	variável	$bizdate	Um parâmetro que controla a criação de partições.
  3	holo_table_name	constante	holo_sale_detail	O nome da tabela particionada do Hologres.
  4	odps_project	constante	hologres_test	O nome do projeto do MaxCompute.
  5	odps_table_name	constante	odps_sale_detail	O nome da tabela particionada do MaxCompute.
  6	partition_key	constante	sale_date	A chave de partição da tabela do MaxCompute.

• Gravar dados particionados em uma tabela temporária

  Esta etapa é um módulo SQL do Hologres. O código SQL é exibido a seguir.

  Nota

  O Hologres V2.1.17 e versões posteriores suportam o Serverless Computing. Para cenários como a importação de dados offline em grande escala, grandes jobs ETL e consultas de alto volume em tabelas estrangeiras, execute essas tarefas com o Serverless Computing. Esse recurso consome recursos serverless adicionais em vez dos recursos da sua instância, o que melhora a estabilidade da instância e reduz a probabilidade de erros de falta de memória (OOM). Você não precisa reservar recursos de computação extras para a sua instância e é cobrado apenas pelas tarefas que executa. Para obter mais informações sobre o Serverless Computing, consulte Serverless Computing. Para obter instruções sobre como usar o Serverless Computing, consulte Use Serverless Computing.

  -- Clean up any potential temporary tables.
  BEGIN ;
  DROP TABLE IF EXISTS ${holo_table_name}_tmp_${datetime1};
  COMMIT ;
  -- Create a temporary table.
  SET hg_experimental_enable_create_table_like_properties=on;
  BEGIN ;
  CALL HG_CREATE_TABLE_LIKE ('${holo_table_name}_tmp_${datetime1}', 'select * from ${holo_table_name}'); 
  COMMIT;
  -- (Optional) We recommend that you use Serverless Computing to run large-scale offline data imports and ETL jobs.
  SET hg_computing_resource = 'serverless';
  -- Insert data into the temporary table.
  INSERT INTO ${holo_table_name}_tmp_${datetime1}
  SELECT *
  FROM public.${odps_table_name}
  WHERE ${partition_key}='${datetime1}';
  -- Reset the configuration to ensure that non-essential SQL queries do not use serverless resources.
  RESET hg_computing_resource;

  Vincule os parâmetros básicos como uma entrada upstream para este módulo. Na configuração Node Context desse nó, a seção Input Parameters of This Node contém cinco parâmetros: datetime1, holo_table_name, odps_project, odps_table_name e partition_key. A origem de valor para todos os parâmetros é o nó upstream holo_doc_test.501890067_out. Adicione-os manualmente. A seção Output Parameters of This Node está vazia.

• Substituir uma tabela filha

  Esta etapa é um módulo SQL do Hologres que substitui uma tabela filha existente. O sistema executa as operações dentro de uma transação para garantir a atomicidade. O código SQL é exibido a seguir.

  -- Replace the child table if it exists.
  BEGIN ;
  -- Delete the existing child table.
  DROP TABLE IF EXISTS ${holo_table_name}_${datetime1};
  -- Rename the temporary table.
  ALTER TABLE ${holo_table_name}_tmp_${datetime1} RENAME TO ${holo_table_name}_${datetime1};
  -- Attach the temporary table to the specified partitioned table.
  ALTER TABLE ${holo_table_name} ATTACH PARTITION ${holo_table_name}_${datetime1}
  FOR VALUES IN ('${datetime1}');
  COMMIT ;

  Vincule os parâmetros básicos como uma entrada upstream para este módulo. Em Node Context, na seção Input Parameters of This Node, adicione manualmente os parâmetros datetime1 e holo_table_name. Defina as fontes de valor para as saídas do nó upstream hologres_doc_test.502282431_out:datetime1 e hologres_doc_test.502282431_out:holo_table_name, respectivamente.

• Coletar estatísticas para a tabela particionada

  Esta etapa é um módulo SQL do Hologres que coleta estatísticas para a tabela pai. O código SQL é exibido a seguir.

  -- Run ANALYZE on the parent table of the partitioned table after a large data import.
  ANALYZE ${holo_table_name};

  Vincule os parâmetros básicos como uma entrada upstream para este módulo. Em Node Context, na seção Input Parameters of This Node, clique em Add e adicione manualmente o parâmetro holo_table_name. Defina a origem de valor como holo_doc_test.501890067_out:holo_table_name.

• Limpar tabelas filhas expiradas

  Em ambientes de produção, os dados possuem um ciclo de vida e é necessário limpar as partições que ultrapassaram o período de retenção.

  Este exemplo mostra como armazenar apenas as partições dos últimos 31 dias no Hologres. Como o parâmetro está definido como datepre31=${yyyymmdd-31}, o código SQL para excluir tabelas filhas expiradas é exibido a seguir. Antes de excluir uma tabela filha de uma tabela particionada física, primeiro desanexe-a da tabela pai (DETACH PARTITION) e, em seguida, execute o DROP TABLE. Caso contrário, a tabela filha permanecerá anexada à tabela pai.

  -- Clean up expired child tables: Detach the child partition first, and then drop the table.
  BEGIN ;
  ALTER TABLE ${holo_table_name} DETACH PARTITION ${holo_table_name}_${datepre31};
  DROP TABLE IF EXISTS ${holo_table_name}_${datepre31};
  COMMIT ;

  Por exemplo, se o bizdate for 20200309, o datepre31 será 20200207. Isso define corretamente a data para a limpeza da partição.

  Também é necessário vincular os parâmetros básicos como uma entrada upstream para este módulo. No painel Node Context, na seção Input Parameters of This Node, clique em Add e adicione manualmente dois parâmetros: o nome do parâmetro datepre31 com a origem de valor definida como a saída do nó upstream holo_doc_test.501890067_out:datepre31; e o nome do parâmetro holo_table_name com a origem de valor definida como a saída do nó upstream holo_doc_test.501890067_out:holo_table_name.

Importar um job do DataWorks com o assistente de migração

• Devido à complexidade do job, utilize o assistente de migração do DataWorks para importar o pacote de job fornecido para o seu projeto. Isso oferece um job pré-configurado do DataWorks descrito neste tópico, que pode ser personalizado ao alterar os parâmetros ou scripts para atender às necessidades específicas da sua empresa.

• Baixe o seguinte pacote de job: DataWorks job package.

1. Abra o assistente de migração do DataWorks. Para mais informações, consulte Open the migration assistant.

2. No painel de navegação à esquerda do assistente de migração, clique em Migration in DataWorks > DataWorks import.

3. Na página DataWorks import, clique em Create Import Task no canto superior direito.

4. Na caixa de diálogo Create Import Task, configure os parâmetros.

   Parâmetro	Descrição
   Import Package Name	Um nome personalizado. O nome de importação pode conter letras maiúsculas, letras minúsculas, caracteres chineses, dígitos, sublinhados (_) e pontos (.).
   Upload Method	O método de upload do arquivo. Upload Local File: Use este método para fazer o upload de um arquivo de pacote de exportação de 30 MB ou menos no workspace do DataWorks. OSS Endpoint: Se um arquivo de pacote de exportação for maior que 30 MB, faça o upload do arquivo para o OSS e, em seguida, insira a URL do arquivo no seu workspace do DataWorks. Para obter a URL do arquivo, acesse a página de detalhes do arquivo no console do OSS, encontre a seção File Url e copie a URL. Para obter mais informações sobre como fazer upload de arquivos, consulte Upload files. Para obter mais informações sobre como obter um link de download do OSS, consulte Share files.
   Remarks	Uma breve descrição da tarefa de importação.

5. Clique em OK para acessar a página Import Task Settings e configure os mapeamentos.

6. Clique em Start Import no canto inferior esquerdo e, na caixa de diálogo Confirm, clique em OK.

   1. Após a importação ser bem-sucedida, o job agendado descrito acima é exibido no seu módulo Data Development.

   2. As instruções DDL relacionadas também aparecem no seu fluxo de trabalho manual.