AnalyticDB:Import MaxCompute data by using external tables

Importação regular

1. Abra o editor SQL. Faça logon no console do AnalyticDB for MySQL. No canto superior esquerdo, selecione uma região. No painel de navegação à esquerda, clique em Clusters, localize e clique no ID do cluster e escolha Job Development > SQL Development.

2. Crie um banco de dados externo.

   CREATE EXTERNAL DATABASE adb_external_db;

3. Crie uma tabela externa que mapeie a tabela de origem do MaxCompute.

   Nota

   • As colunas da tabela externa devem corresponder exatamente às da tabela de origem do MaxCompute: mesmos nomes de campos, mesma quantidade de campos, mesma ordem de campos e tipos de dados compatíveis. Para todos os parâmetros de TABLE_PROPERTIES, consulte CREATE EXTERNAL TABLE.

   Parâmetro	Descrição
   ENGINE='ODPS'	Especifica o MaxCompute como mecanismo de armazenamento.
   endpoint	O endpoint de VPC do MaxCompute. Apenas endpoints de VPC têm suporte. Consulte Endpoints de VPC para obter os endpoints por região.
   accessid	O AccessKey ID com permissões de leitura no projeto do MaxCompute.
   accesskey	O AccessKey secret correspondente ao AccessKey ID.
   partition_column	O nome da coluna de partição. Omita este parâmetro se a tabela de origem do MaxCompute não for particionada.
   project_name	O nome do workspace do MaxCompute.
   table_name	O nome da tabela de origem no MaxCompute.

   CREATE EXTERNAL TABLE IF NOT EXISTS adb_external_db.test_adb (
       id int,
       name varchar(1023),
       age int,
       dt string
   ) ENGINE='ODPS'
   TABLE_PROPERTIES='{
   "accessid":"yourAccessKeyID",
   "endpoint":"http://service.cn-hangzhou.maxcompute.aliyun.com/api",
   "accesskey":"yourAccessKeySecret",
   "partition_column":"dt",
   "project_name":"odps_project",
   "table_name":"odps_nopart_import_test"
   }';

   Principais parâmetros:

4. Verifique a tabela externa consultando os dados.

   SELECT * FROM adb_external_db.test_adb;

   Saída esperada:

   +------+-------+------+---------+
   | id   | name  | age  |   dt    |
   +------+-------+------+---------+
   |    1 | james |   10 |  202207 |
   |    2 | bond  |   20 |  202207 |
   |    3 | jack  |   30 |  202207 |
   |    4 | lucy  |   40 |  202207 |
   +------+-------+------+---------+

5. Crie um banco de dados e uma tabela de destino no AnalyticDB for MySQL.

   Nota

   A tabela de destino e a tabela externa devem ter a mesma ordem e quantidade de campos, com tipos de dados compatíveis.

   CREATE DATABASE adb_demo;

   CREATE TABLE IF NOT EXISTS adb_demo.adb_import_test(
       id int,
       name string,
       age int,
       dt string,
       PRIMARY KEY(id,dt)
   )
   DISTRIBUTED BY HASH(id)
   PARTITION BY VALUE('dt');

6. Importe os dados. Escolha um dos seguintes métodos:

   • **Método 1 — INSERT INTO:** Importa dados e ignora linhas com chaves primárias duplicadas (equivalente a INSERT IGNORE INTO). Consulte INSERT INTO.

     INSERT INTO adb_demo.adb_import_test
     SELECT * FROM adb_external_db.test_adb;

     Para importar apenas de uma partição específica:

     INSERT INTO adb_demo.adb_import_test
     SELECT * FROM adb_external_db.test_adb
     WHERE dt = '202207';

   • **Método 2 — INSERT OVERWRITE INTO:** Sobrescreve todos os dados existentes na tabela de destino.

     INSERT OVERWRITE INTO adb_demo.adb_import_test
     SELECT * FROM adb_external_db.test_adb;

   • **Método 3 — INSERT OVERWRITE INTO assíncrono:** Envia a importação como um job em segundo plano. Adicione a dica /*+ direct_batch_load=true*/ para acelerar o job. Consulte Gravações assíncronas.

     SUBMIT JOB
     INSERT OVERWRITE INTO adb_demo.adb_import_test
     SELECT * FROM adb_external_db.test_adb;

     O comando retorna um ID de job:

     +---------------------------------------+
     | job_id                                |
     +---------------------------------------+
     | 2020112122202917203100908203303****** |
     +---------------------------------------+

     Use o ID do job para monitorá-lo ou cancelá-lo. Consulte Enviar uma tarefa de importação assíncrona.

Importação elástica

A importação elástica executa a importação dentro de um job Serverless Spark, reduzindo o impacto nas cargas de trabalho em tempo real. Esse método requer Milvus V3.1.10.0 ou posterior e um grupo de recursos de jobs configurado.

1. Abra o editor SQL. Faça logon no console do AnalyticDB for MySQL. No canto superior esquerdo, selecione uma região. No painel de navegação à esquerda, clique em Clusters, localize e clique no ID do cluster e escolha Job Development > SQL Development.

2. Crie um banco de dados (ignore esta etapa se já existir um).

   CREATE DATABASE adb_demo;

3. Crie uma tabela externa.

   Nota

   • O nome da tabela externa deve corresponder ao nome do projeto do MaxCompute. Caso contrário, a criação da tabela externa falhará.

   • A importação elástica exige a criação de tabelas externas com CREATE TABLE (não use CREATE EXTERNAL TABLE).

   • As colunas da tabela externa devem corresponder exatamente às da tabela de origem do MaxCompute: mesmos nomes de campos, mesma quantidade de campos, mesma ordem de campos e mesmos tipos de dados.

   CREATE TABLE IF NOT EXISTS test_adb
   (
       id int,
       name string,
       age int,
       dt string
   )
   ENGINE='ODPS'
   TABLE_PROPERTIES='{
   "endpoint":"http://service.cn-hangzhou.maxcompute.aliyun-inc.com/api",
   "accessid":"yourAccessKeyID",
   "accesskey":"yourAccessKeySecret",
   "partition_column":"dt",
   "project_name":"odps_project",
   "table_name":"odps_nopart_import_test"
   }';

   Para descrições dos parâmetros de TABLE_PROPERTIES, consulte Descrições de parâmetros.

4. Verifique a tabela externa consultando os dados.

   SELECT * FROM adb_demo.test_adb;

5. Crie a tabela de destino no AnalyticDB for MySQL.

   Nota

   A tabela de destino e a tabela externa devem ter os mesmos nomes de campos, quantidade de campos, ordem de campos e tipos de dados.

   CREATE TABLE IF NOT EXISTS adb_import_test
   (
       id int,
       name string,
       age int,
       dt string,
       PRIMARY KEY(id,dt)
   )
   DISTRIBUTED BY HASH(id)
   PARTITION BY VALUE('dt') LIFECYCLE 30;

6. Importe os dados usando um dos métodos abaixo. Ambos usam os parâmetros de dica elastic_load e elastic_load_configs para ativar e configurar o job de importação elástica. Defina elastic_load como true para ativar a importação elástica (padrão: false). Use elastic_load_configs para especificar parâmetros de configuração entre [ ]; separe múltiplos parâmetros com |. elastic_load_configs aceita os seguintes parâmetros:

   • Método 1 — Importação elástica síncrona:

     /*+ elastic_load=true, elastic_load_configs=[adb.load.resource.group.name=resource_group]*/
     INSERT OVERWRITE INTO adb_demo.adb_import_test SELECT * FROM adb_demo.test_adb;

   • Método 2 — Importação elástica assíncrona:

     Importante

     Filas de prioridade não têm suporte para tarefas de importação elástica assíncrona.

     /*+ elastic_load=true, elastic_load_configs=[adb.load.resource.group.name=resource_group]*/
     SUBMIT JOB INSERT OVERWRITE INTO adb_demo.adb_import_test SELECT * FROM adb_demo.test_adb;

     O comando retorna um ID de job. Use-o para monitorar ou cancelar o job. Consulte Enviar uma tarefa de importação assíncrona.

   Parâmetro	Obrigatório	Descrição
   adb.load.resource.group.name	Sim	Nome do grupo de recursos de jobs que executa o job de importação elástica.
   adb.load.job.max.acu	Não	Recursos máximos para o job de importação elástica. Unidade: AnalyticDB Compute Units (ACUs). Mínimo: 5 ACUs. Padrão: número de shards mais 1. Para consultar o número de shards: SELECT count(1) FROM information_schema.kepler_meta_shards;
   spark.driver.resourceSpec	Não	Tipo de recurso do driver Spark. Padrão: small. Consulte a coluna Type na tabela de Especificações de recursos do Spark.
   spark.executor.resourceSpec	Não	Tipo de recurso do executor Spark. Padrão: large. Consulte a coluna Type na tabela de Especificações de recursos do Spark.
   spark.adb.executorDiskSize	Não	Capacidade de disco do executor Spark. Valores válidos: (0, 100]. Unidade: GiB. Padrão: 10 GiB. Consulte Especificar recursos de driver e executor.

7. (Opcional) Verifique se o job foi executado como uma tarefa de importação elástica.

   SELECT job_name, (job_type = 3) AS is_elastic_load
   FROM INFORMATION_SCHEMA.kepler_meta_async_jobs
   WHERE job_name = "2023081818010602101701907303151******";

   Saída esperada:

   +---------------------------------------+------------------+
   | job_name                              | is_elastic_load  |
   +---------------------------------------+------------------+
   | 2023081517195203101701907203151****** |       1          |
   +---------------------------------------+------------------+

   is_elastic_load = 1 indica uma tarefa de importação elástica. is_elastic_load = 0 indica uma tarefa de importação regular.