ApsaraDB for SelectDB:Importer des données avec OSS Load

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", ...
);

Description des données (data_desc)

Chaque bloc data_desc décrit un ensemble de fichiers à importer. La syntaxe complète est la suivante :

[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", ...)]

Paramètres de connexion OSS (WITH S3)

Propriétés de la tâche (PROPERTIES)

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");

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

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"
);

Vérifier l'état d'une tâche

OSS Load fonctionne de manière asynchrone. Après avoir soumis la tâche, utilisez la commande SHOW LOAD pour suivre sa progression.

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

Exemples :

Interrogez les tâches de la base example_db dont le libellé correspond à 2014_01_02, en retournant les 10 plus anciennes :

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

Recherchez une tâche spécifique et triez les résultats par heure de début :

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

Listez les tâches actuellement à l'état LOADING :

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

Effectuez une requête paginée (ignorez les 5 premiers résultats, retournez les 10 suivants) :

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;

Annuler une tâche d'importation

Annulez toute tâche n'ayant pas encore atteint l'état FINISHED ou CANCELLED. Après l'annulation, toutes les données écrites par la tâche font l'objet d'une restauration (rollback).

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

Exemples :

Annulez une tâche spécifique :

CANCEL LOAD
FROM example_db
WHERE LABEL = "example_db_test_load_label";

Annulez toutes les tâches dont le libellé commence par example_ :

CANCEL LOAD
FROM example_db
WHERE LABEL LIKE "example_";

Expiration du délai d'importation

Le délai d'expiration par défaut est de 4 heures (14 400 secondes). Si une tâche dépasse cette limite, évitez de simplement augmenter le paramètre timeout : les nouvelles tentatives longues deviennent coûteuses lorsqu'une tâche échoue tardivement.

Privilégiez plutôt le découpage des fichiers volumineux en fichiers plus petits, puis lancez plusieurs tâches d'importation distinctes.

Si le volume de données reste raisonnable mais que la tâche expire tout de même, augmentez la valeur de load_parallelism afin d'exécuter davantage de tâches en parallèle, puis ajustez le paramètre timeout en conséquence.