Use OSS to import data to an ApsaraDB for SelectDB instance - ApsaraDB for SelectDB: Central de Documentação do Alibaba Cloud

Quando usar o OSS Load

O OSS Load é a escolha ideal quando:

Você precisa importar grandes lotes de arquivos (de dezenas a centenas de gigabytes)
Os arquivos estão armazenados em um bucket do OSS na mesma região da instância do SelectDB
É necessário carregar arquivos CSV, PARQUET ou ORC

O OSS Load utiliza internamente o protocolo S3; portanto, a sintaxe contém as palavras-chave AWS e S3. Esse comportamento é esperado, pois o ApsaraDB for SelectDB suporta qualquer armazenamento de objetos compatível com S3, e o acesso ao OSS ocorre via endpoint compatível com S3.

Pré-requisitos

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

Um par de AccessKey. Consulte Criar um par de AccessKey.
Um bucket do OSS na mesma região da instância do ApsaraDB for SelectDB. Consulte Introdução ao uso do console do OSS.
Acesso ao OSS via Virtual Private Cloud (VPC), pois o OSS Load usa o endpoint interno da VPC, não o endpoint público

Sintaxe

LOAD LABEL <load_label>
(
    data_desc1[, data_desc2, ...]
)
WITH S3
(
    "AWS_PROVIDER" = "OSS",
    "AWS_REGION" = "<region>",
    "AWS_ENDPOINT" = "<endpoint>",
    "AWS_ACCESS_KEY" = "<AccessKey ID>",
    "AWS_SECRET_KEY" = "<AccessKey secret>"
)
PROPERTIES
(
    "key1" = "value1", ...
);

Nota

O caminho do bucket do OSS deve começar com s3://.

Parâmetros

Descrição dos dados (`data_desc`)

Cada bloco data_desc descreve um grupo de arquivos para importação. A sintaxe completa é:

[MERGE|APPEND|DELETE]
DATA INFILE
(
    "file_path1"[, "file_path2", ...]
)
[NEGATIVE]
INTO TABLE `<table_name>`
[PARTITION (p1, p2, ...)]
[COLUMNS TERMINATED BY "<column_separator>"]
[FORMAT AS "<file_type>"]
[(column_list)]
[COLUMNS FROM PATH AS (c1, c2, ...)]
[PRECEDING FILTER predicate]
[SET (column_mapping)]
[WHERE predicate]
[DELETE ON expr]
[ORDER BY source_sequence]
[PROPERTIES ("key1"="value1", ...)]

Parâmetro	Descrição
`MERGE	APPEND	DELETE`	Tipo de mesclagem de dados. Padrão: `APPEND`. As opções `MERGE` e `DELETE` aplicam-se apenas a tabelas no modelo Unique Key. Ao definir como `MERGE`, use `DELETE ON` para especificar a coluna delete_flag. Se definido como `DELETE`, as linhas de dados importadas serão excluídas da tabela.
`DATA INFILE`	Caminho do(s) arquivo(s) para importar. Suporta caracteres curinga. Deve ser um caminho de arquivo, não de diretório.
`NEGATIVE`	Importa dados usando um método negativo, que inverte valores inteiros em colunas agregadas por SUM para compensar dados incorretos importados anteriormente. Aplica-se apenas a tabelas que usam agregação SUM com colunas inteiras.
`PARTITION (p1, p2, ...)`	Partições de destino da importação. Dados fora das partições especificadas são ignorados.
`COLUMNS TERMINATED BY`	Delimitador de coluna. Válido apenas para arquivos CSV. Deve ter um único byte.
`FORMAT AS`	Formato do arquivo. Valores válidos: `CSV`, `PARQUET`, `ORC`. Padrão: `CSV`.
`column_list`	Ordem das colunas no arquivo de origem. Consulte Converter dados de origem.
`COLUMNS FROM PATH AS`	Colunas a extrair do caminho do arquivo.
`PRECEDING FILTER predicate`	Condições predefinidas para filtragem de dados. Os dados são mesclados sequencialmente às linhas de dados de origem com base nos valores dos parâmetros `column list` e `COLUMNS FROM PATH AS`. Em seguida, a filtragem ocorre conforme as condições predefinidas.
`SET (column_mapping)`	Funções de transformação de colunas aplicadas aos dados de origem.
`WHERE predicate`	Condições usadas para filtrar os dados importados.
`DELETE ON expr`	Uso exclusivo com o tipo `MERGE`. Especifica a coluna delete_flag e a condição correspondente. Aplica-se apenas a tabelas no modelo Unique Key.
`ORDER BY`	Coluna de sequência para a importação. Aplica-se apenas a tabelas no modelo Unique Key, garantindo a ordenação dos dados durante a importação.
`PROPERTIES`	Parâmetros adicionais de formato. Para arquivos JSON, defina `json_root`, `jsonpaths` e `fuzzy_parse`.

Parâmetros de conexão do OSS (`WITH S3`)

Parâmetro	Descrição
`AWS_PROVIDER`	Provedor de armazenamento de objetos. Defina como `OSS`.
`AWS_REGION`	Região onde o bucket do OSS está localizado.
`AWS_ENDPOINT`	Endpoint usado para acessar os dados no OSS. Consulte Regiões e endpoints. O bucket do OSS e a instância do SelectDB devem estar na mesma região.
`AWS_ACCESS_KEY`	AccessKey ID usado para acessar o OSS.
`AWS_SECRET_KEY`	AccessKey secret usado para acessar o OSS.

Propriedades do job (`PROPERTIES`)

Parâmetro	Padrão	Descrição
`timeout`	`14400` (4 horas)	Tempo limite do job de importação, em segundos.
`max_filter_ratio`	`0`	Proporção máxima de linhas filtráveis devido a problemas de qualidade dos dados. Valores válidos: 0 a 1.
`exec_mem_limit`	`2147483648` (2 GiB)	Memória máxima disponível para o job de importação, em bytes.
`strict_mode`	`false`	Indica se o modo estrito deve ser ativado para o job de importação.
`timezone`	`Asia/Shanghai`	Fuso horário para funções relacionadas a tempo, como `strftime`, `alignment_timestamp` e `from_unixtime`. Consulte Lista de todos os fusos horários.
`load_parallelism`	`1`	Número de tarefas de importação simultâneas. Aumente este valor para acelerar importações grandes.
`send_batch_parallelism`	—	Quantidade de tarefas simultâneas para envio de dados em lotes. Limitado pela configuração do BE `max_send_batch_parallelism_per_job`.
`load_to_single_tablet`	`false`	Define se todos os dados devem ser importados para um único tablet por partição. Aplica-se apenas a tabelas no modelo Duplicate Key com particionamento aleatório.

Importar dados do OSS

Este exemplo cria uma tabela, prepara um arquivo CSV de amostra no OSS e o carrega usando o OSS Load.

Etapa 1: Crie a tabela de destino.

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

Etapa 2: Envie seu arquivo de dados para o OSS.

Faça upload de um arquivo CSV chamado test_file.txt para o seu bucket do OSS. O conteúdo do arquivo deve ser semelhante a:

1,yang,32,shanghai,http://example.com
2,wang,22,beijing,http://example.com
3,xiao,23,shenzhen,http://example.com

Etapa 3: Envie o job de importação.

LOAD LABEL test_db.test_label_1
(
    DATA INFILE("s3://your_bucket_name/test_file.txt")
    INTO TABLE test_table
    COLUMNS TERMINATED BY ","
)
WITH S3
(
    "AWS_PROVIDER" = "OSS",
    "AWS_REGION" = "oss-cn-beijing",
    "AWS_ENDPOINT" = "oss-cn-beijing-internal.aliyuncs.com",
    "AWS_ACCESS_KEY" = "<your_access_key>",
    "AWS_SECRET_KEY" = "<your_secret_key>"
)
PROPERTIES
(
    "timeout" = "3600"
);

A instrução LOAD possui quatro seções:

LABEL: Identificador exclusivo deste job de importação, usado para consultar seu status e cancelá-lo, se necessário.
Declaração de dados: Caminho do arquivo de origem, formato e tabela de destino.
WITH S3: Credenciais de conexão e endpoint do OSS.
PROPERTIES: Configurações no nível do job, como tempo limite.

Monitorar e gerenciar jobs de importação

Verificar o status do job

O OSS Load opera de forma assíncrona. Após enviar o job, use SHOW LOAD para acompanhar seu progresso.

SHOW LOAD
[FROM <db_name>]
[
   WHERE
   [LABEL [ = "your_label" | LIKE "label_matcher"]]
   [STATE = ["PENDING"|"ETL"|"LOADING"|"FINISHED"|"CANCELLED"]]
]
[ORDER BY ...]
[LIMIT limit][OFFSET offset];

Parâmetro	Padrão	Descrição
`db_name`	Banco de dados atual	Banco de dados a ser consultado.
`LABEL`	—	Filtra por rótulo. Suporta correspondência exata (`=`) e correspondência por padrão (`LIKE`).
`STATE`	—	Filtra pelo estado do job.
`ORDER BY`	—	Ordena os resultados.
`LIMIT`	Todos os registros	Número máximo de registros a retornar.
`OFFSET`	`0`	Quantidade de registros a pular.

Exemplos:

Consulte jobs em example_db com rótulos correspondentes a 2014_01_02, retornando os 10 mais antigos:

SHOW LOAD FROM example_db WHERE LABEL LIKE "2014_01_02" LIMIT 10;

Consulte um job específico e ordene pelo horário de início:

SHOW LOAD FROM example_db WHERE LABEL = "load_example_db_20140102" ORDER BY LoadStartTime DESC;

Consulte jobs atualmente no estado LOADING:

SHOW LOAD FROM example_db WHERE LABEL = "load_example_db_20140102" AND STATE = "LOADING";

Consulte com paginação (pule os primeiros 5 e retorne os próximos 10):

SHOW LOAD FROM example_db ORDER BY LoadStartTime DESC LIMIT 5,10;
SHOW LOAD FROM example_db ORDER BY LoadStartTime DESC LIMIT 10 OFFSET 5;

Cancelar um job de importação

Cancele jobs que ainda não estejam no estado FINISHED ou CANCELLED. Após o cancelamento, quaisquer dados gravados pelo job sofrem rollback.

CANCEL LOAD
[FROM <db_name>]
WHERE [LABEL = "<load_label>" | LABEL LIKE "<label_pattern>"];

Parâmetro	Padrão	Descrição
`db_name`	Banco de dados atual	Banco de dados que contém o job de importação.
`load_label`	—	Rótulo do job a ser cancelado. Suporta correspondência exata e correspondência por padrão com `LABEL LIKE`.

Exemplos:

Cancele um job específico:

CANCEL LOAD
FROM example_db
WHERE LABEL = "example_db_test_load_label";

Cancele todos os jobs cujos rótulos comecem com example_:

CANCEL LOAD
FROM example_db
WHERE LABEL LIKE "example_";

Solução de problemas

A importação atinge o tempo limite

O tempo limite padrão é de 4 horas (14.400 segundos). Caso um job ultrapasse esse limite, evite simplesmente aumentar o timeout, pois novas tentativas longas tornam-se custosas quando o job falha tardiamente.

Em vez disso, divida arquivos grandes em menores e importe-os em múltiplos jobs.

Se o volume de dados estiver dentro do esperado, mas o job ainda assim atingir o tempo limite, aumente o valor de load_parallelism para executar mais tarefas simultaneamente e ajuste o parâmetro timeout adequadamente.