ApsaraDB for SelectDB:Usar INSERT INTO para importar dados

Executar ETL em uma tabela interna

Para transformar dados de uma tabela do SelectDB e gravar os resultados em outra:

INSERT INTO bj_store_sales
SELECT id, total, user_id, sale_timestamp FROM store_sales WHERE region = "bj";

Este comando lê linhas da tabela store_sales onde region = "bj" e as grava em bj_store_sales.

Sincronizar dados de um data lake

Os catalogs do SelectDB permitem mapear fontes de dados externas — incluindo Hive, Iceberg, Hudi, Elasticsearch e fontes Java Database Connectivity (JDBC) — e consultá-las com consultas federadas. Use um catalog para sincronizar dados de um data lake para uma tabela do SelectDB.

O exemplo a seguir sincroniza dados de uma fonte Hive para o SelectDB.

1. Conecte-se à sua instância do SelectDB. Para mais detalhes, consulte Conectar-se a uma instância do ApsaraDB for SelectDB usando um cliente MySQL.

2. Crie um catalog para integrar a source de dados Hive. Para mais detalhes, consulte Fonte de dados Hive.

3. (Opcional) Crie um banco de dados de destino. Ignore esta etapa se o banco de dados já existir.

   CREATE DATABASE hive_db;

4. Mude para o banco de dados de destino.

   USE hive_db;

5. Crie uma tabela de destino. Se a tabela já existir, verifique se os tipos de coluna correspondem aos da tabela de origem. Para referência de mapeamento de tipos, consulte Mapeamentos de tipos de dados de colunas.

   CREATE TABLE test_Hive2SelectDB
   (
       id int,
       name varchar(50),
       age int
   )
   DISTRIBUTED BY HASH(id) BUCKETS 4
   PROPERTIES("replication_num" = "1");

6. (Opcional) Visualize a tabela antes da importação.

   SELECT * FROM test_Hive2SelectDB;

   

7. Execute a instrução INSERT INTO...SELECT para sincronizar os dados. Atribua um label exclusivo ao job de importação com WITH LABEL.

   INSERT INTO test_Hive2SelectDB WITH LABEL test_label SELECT * FROM hive_catalog.testdb.hive_t;

8. Consulte a tabela de destino para verificar os dados. Os dados da tabela de destino aparecem à esquerda e os dados de origem à direita.

   

CREATE TABLE test_table
(
    id int,
    name varchar(50),
    age int
)
DISTRIBUTED BY HASH(id) BUCKETS 4
PROPERTIES("replication_num" = "1");

Usando um cliente SQL

Agrupe várias instruções INSERT INTO em uma transação para combiná-las em uma única importação:

BEGIN;
INSERT INTO test_table VALUES (1, 'Zhang San', 32),(2, 'Li Si', 45),(3, 'Zhao Liu', 23);
INSERT INTO test_table VALUES (4, 'Wang Yi', 32),(5, 'Zhao Er', 45),(6, 'Li Er', 23);
INSERT INTO test_table VALUES (7, 'Li Yi', 32),(8, 'Wang San', 45),(9, 'Zhao Si', 23);
COMMIT;

Usando uma aplicação JDBC

O exemplo a seguir agrupa várias instruções INSERT INTO em uma única transação usando JDBC. Substitua os valores de espaço reservado pelos seus próprios valores.

public static void main(String[] args) throws Exception {
    // Number of INSERT statements per batch
    int insertNum = 10;
    // Number of rows per INSERT statement
    int batchSize = 10000;

    // Replace <host> and <port> with your VPC (virtual private cloud) endpoint values.
    // Find these on the Instance Details page under Network Information.
    String URL = "jdbc:mysql://<host>:<port>/test_db?useLocalSessionState=true";
    Connection connection = DriverManager.getConnection(URL, "admin", "<password>");
    Statement statement = connection.createStatement();
    statement.execute("BEGIN;");

    for (int num = 0; num < insertNum; num++) {
        StringBuilder sql = new StringBuilder();
        sql.append("INSERT INTO test_table VALUES ");
        for (int i = 0; i < batchSize; i++) {
            if (i > 0) {
                sql.append(",");
            }
            // Replace with your actual field values
            sql.append("(1, 'Zhang San', 32)");
        }
        statement.addBatch(sql.toString());
    }
    statement.addBatch("COMMIT;");
    statement.executeBatch();

    statement.close();
    connection.close();
}

Importação bem-sucedida sem linhas

Se a cláusula SELECT não retornar linhas, o SelectDB exibe:

INSERT INTO tbl1 SELECT * FROM empty_tbl;
Query OK, 0 rows affected (0.02 sec)

Query OK indica que a instrução foi executada sem erros. 0 rows affected significa que nenhum dado foi importado.

Importação bem-sucedida com linhas

INSERT INTO tbl1 SELECT * FROM tbl2;
Query OK, 4 rows affected (0.38 sec)
{'label':'insert_8510c568-9eda-****-9e36-6adc7d35291c', 'status':'visible', 'txnId':'4005'}

A resposta JSON inclui:

Se o status for committed, os dados eventualmente ficarão visíveis. Para verificar:

SHOW TRANSACTION WHERE id=4005;

Se TransactionStatus mostrar visible, os dados estarão disponíveis para consulta.

Importação bem-sucedida com linhas filtradas

Se algumas linhas forem filtradas, o resultado exibirá uma contagem de avisos:

Query OK, 2 rows affected, 2 warnings (0.31 sec)
{'label':'insert_f0747f0e-7a35-****-affa-13a235f4020d', 'status':'visible', 'txnId':'4005'}

Para inspecionar as linhas filtradas, localize o label na saída do SHOW LOAD:

SHOW LOAD WHERE label="insert_f0747f0e-7a35-****-affa-13a235f4020d";

Em seguida, consulte os detalhes do erro usando a URL fornecida na saída:

SHOW LOAD WARNINGS ON "<error-url>";

Importação com falha

Se a importação falhar, nenhum dado será gravado e o SelectDB retornará um erro:

INSERT INTO tbl1 SELECT * FROM tbl2 WHERE k1 = "a";
ERROR 1064 (HY000): all partitions have no load data. url: http://10.74.167.16:8042/api/_load_error_log?file=__shard_2/error_log_insert_stmt_ba8bb9e158e4879-ae8de8507c0bf8a2_ba8bb9e158e4879_ae8de8507c0bf8a2

Obtenha informações detalhadas sobre o erro usando a URL presente na mensagem de erro:

SHOW LOAD WARNINGS ON "<error-url>";

Variáveis de sessão

Atualizações parciais de colunas

Por padrão, o INSERT INTO grava linhas completas. Para atualizar apenas colunas específicas em uma tabela de modelo Unique Key que usa Merge on Write (MOW):

SET enable_unique_key_partial_update = true;

Para obter a lista completa de variáveis, consulte Gerenciamento de variáveis.