prise en charge des fonctionnalités pour la synchronisation des données StarRocks - DataWorks

Versions prises en charge

Toutes les versions d'EMR Serverless StarRocks sont prises en charge.
EMR on ECS est pris en charge : StarRocks version 2.1.
StarRocks Community Edition est pris en charge.
Remarque
- DataWorks se connecte à StarRocks uniquement via un VPC. Vous devez donc déployer StarRocks Community Edition sur un cluster EMR on ECS.
- StarRocks Community Edition est une plateforme ouverte. En cas de problèmes de compatibilité, soumettez un ticket au support technique.

Limitations

Pour la synchronisation en temps réel de base de données complète de MySQL vers StarRocks, la table StarRocks de destination doit utiliser un modèle de clé primaire.
La synchronisation en temps réel de base de données complète de MySQL vers StarRocks ne prend pas en charge les opérations DDL (Data Definition Language), à l'exception de TRUNCATE. Pour les autres opérations DDL, choisissez de les ignorer ou configurez la tâche pour signaler une erreur.

Types de données pris en charge

Seuls les champs numériques, de type chaîne et de type date sont pris en charge.

Connectivité réseau

EMR Serverless StarRocks

Pour garantir la connectivité réseau, ajoutez les adresses IP de votre groupe de ressources DataWorks à la liste d'autorisation d'adresses IP internes de l'instance EMR Serverless StarRocks.

Pour obtenir l'adresse IP du groupe de ressources DataWorks, consultez la rubrique Common configurations: Add a whitelist.
Pour configurer les listes d'autorisation de l'instance EMR Serverless StarRocks :

Sur la page des détails de l'instance, dans la section Basic Information, cliquez sur le lien internal IP address whitelist situé à côté de security group ID pour configurer la liste d'autorisation d'adresses IP internes. Dans la section FE Details, cliquez sur le lien public whitelist situé à côté de Public Endpoint pour configurer la liste d'autorisation publique.

Self-managed StarRocks

Assurez-vous que le groupe de ressources DataWorks peut accéder au query port, au FE port et au BE port de votre instance StarRocks. Il s'agit généralement des ports 9030, 8030 et 8040.

Ajouter une source de données

Avant de développer une tâche de synchronisation dans DataWorks, ajoutez la source de données requise à DataWorks en suivant les instructions de la rubrique Data source configuration. Consultez les descriptions des paramètres dans la console DataWorks pour comprendre leur signification lors de l'ajout d'une source de données.

Sélectionnez un mode de connexion pour StarRocks en fonction de votre environnement réseau :

Scénario 1 : Connexion VPC

Une connexion VPC offre une faible latence et une sécurité élevée, sans nécessiter d'accès au réseau public.

Cas d'utilisation : Votre instance StarRocks et le groupe de ressources serverless se trouvent dans le même VPC.
Modes pris en charge : Mode instance Alibaba Cloud et mode chaîne de connexion :
- Sélectionnez ApsaraDB for RDS : Choisissez directement l'instance StarRocks dans le même VPC. Le système récupère automatiquement les informations de connexion ; aucune configuration manuelle n'est requise.
- Sélectionnez User-created Data Store with Public IP Addresses : Saisissez manuellement le point de terminaison intranet, l'adresse IP, le port et l'URL de chargement (Load URL) de l'instance.

Scénario 2 : Connexion Internet

La transmission de données via Internet présente des risques de sécurité. Utilisez des contrôles de sécurité tels que les listes d'autorisation et le contrôle d'accès basé sur l'IP.

Cas d'utilisation : Vous devez accéder à une instance StarRocks via Internet, par exemple entre différentes régions ou depuis un environnement local.
Mode pris en charge : Mode chaîne de connexion (assurez-vous que l'accès au réseau public est activé pour l'instance StarRocks) :
- Sélectionnez User-created Data Store with Public IP Addresses : Saisissez manuellement le point de terminaison public, l'adresse IP, le port et l'URL de chargement (Load URL) de l'instance.

Remarque

Par défaut, les groupes de ressources serverless ne peuvent pas accéder à Internet. Pour vous connecter à une instance StarRocks via un point de terminaison public, configurez une passerelle NAT et une EIP pour le VPC associé. Assurez-vous également que le groupe de ressources peut accéder au query port, au FE port et au BE port de l'instance StarRocks, qui sont généralement les ports 9030, 8030 et 8040.

Si vous utilisez EMR Serverless StarRocks, définissez Host Address/IP Address sur Internal Endpoint ou Public network address, et utilisez le query port pour le port.

FE : Ces informations figurent sur la page des détails de l'instance.

Dans la section FE Details, recherchez le public endpoint et le query port (la valeur par défaut est 9030).
Database : Après vous être connecté à l'instance à l'aide du gestionnaire EMR StarRocks Manager, vous pouvez trouver la base de données dans la vue SQL Editor ou Metadata Management.

Remarque
Pour créer une base de données, exécutez directement des instructions SQL dans l'éditeur SQL.

Tâches de synchronisation des données

Pour plus d'informations sur le point d'entrée et la procédure de configuration d'une tâche de synchronisation, consultez les guides de configuration suivants.

Synchronisation par lots de table unique

Sources prises en charge : Toutes les sources de données prises en charge par Data Integration.
Pour plus d'informations, consultez les rubriques Codeless UI et Code Editor.
Pour obtenir la liste complète des paramètres et des exemples de scripts pour l'éditeur de code, consultez la rubrique Annexe : Exemples de scripts et descriptions des paramètres.

Synchronisation en temps réel de table unique

Source prise en charge : Kafka
Guide de configuration : Configure a real-time data synchronization task

Synchronisation par lots de base de données complète

Source prise en charge : MySQL
Guide de configuration : Configure a batch data synchronization task

Synchronisation en temps réel de base de données complète

Sources prises en charge : MySQL, Oracle et PolarDB
Guide de configuration : Configure a real-time data synchronization task

Exemples de scripts et paramètres

Configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code

Pour configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code, définissez les paramètres associés dans le script selon les exigences de format de script unifié. Pour plus d'informations, consultez la rubrique Script mode configuration. Les informations suivantes décrivent les paramètres à configurer pour les sources de données lors de la configuration d'une tâche de synchronisation par lots à l'aide de l'éditeur de code.

Exemple de script Reader

{
    "stepType": "starrocks",
    "parameter": {
        "selectedDatabase": "didb1",
        "datasource": "starrocks_datasource",
        "column": [
            "id",
            "name"
        ],
        "where": "id>100",
        "table": "table1",
        "splitPk": "id"
    },
    "name": "Reader",
    "category": "reader"
}

Paramètres du script Reader

Parameter	Description	Required	Default
datasource	Le nom de la source de données StarRocks.	Oui	Aucun
selectedDatabase	Le nom de la base de données StarRocks.	Non	Le nom de la base de données configuré pour la source de données StarRocks.
column	Les colonnes de la table source à synchroniser. Pour ajouter une indication SET_VAR lors de la lecture des données depuis StarRocks, ajoutez l'indication avant le premier nom de colonne dans le tableau column. Par exemple, pour synchroniser la colonne `id` avec l'indication `SET_VAR(enable_spill = true)`, définissez le paramètre column sur `[ "/+ SET_VAR(enable_spill = true)/ id"]`.	Oui	Aucun
where	La condition de filtre. Par exemple, pour synchroniser uniquement les données générées le jour actuel, définissez la condition where sur `gmt_create>${bizdate}`. La condition where permet la synchronisation incrémentielle des données. Si vous ne fournissez pas de clause where ou si vous laissez le paramètre where vide, une synchronisation complète des données est effectuée.	Non	Aucun
table	La table source à partir de laquelle vous souhaitez synchroniser les données.	Oui	Aucun
splitPk	La colonne utilisée pour le partitionnement parallèle des données pendant la synchronisation. La spécification de ce paramètre améliore les performances. Pour de meilleurs résultats, utilisez la clé primaire de la table, car une clé uniformément distribuée aide à prévenir les points chauds de données.	Non	Aucun

Exemple de script Writer

{
    "stepType": "starrocks",
    "parameter": {
        "selectedDatabase": "didb1",
        "loadProps": {
            "row_delimiter": "\\x02",
            "column_separator": "\\x01"
        },
        "datasource": "starrocks_public",
        "column": [
            "id",
            "name"
        ],
        "loadUrl": [
            "1.1.X.X:8030"
        ],
        "table": "table1",
        "preSql": [
            "truncate table table1"
        ],
        "postSql": [
        ],
        "maxBatchRows": 500000,
        "maxBatchSize": 5242880,
        "strategyOnError": "exit"
    },
    "name": "Writer",
    "category": "writer"
}

Paramètres du script Writer

Parameter	Description	Required	Default
datasource	Le nom de la source de données StarRocks.	Oui	Aucun
selectedDatabase	Le nom de la base de données StarRocks.	Non	Le nom de la base de données configuré pour la source de données StarRocks.
loadProps	Remarque Lors de l'écriture dans StarRocks avec Stream Load, la politique d'écriture (Upsert ou Append) est déterminée par le modèle de la table de destination et ne nécessite aucune configuration. Les tables avec un modèle de clé primaire utilisent une politique Upsert, tandis que les autres modèles utilisent par défaut une politique Append. Les paramètres de requête pour StarRocks Stream Load. Lors de l'importation de données CSV, vous pouvez configurer les paramètres suivants. Si aucune configuration spéciale n'est requise, utilisez {} : column_separator : le délimiteur de colonne pour l'importation CSV. Valeur par défaut : \t. row_delimiter : le délimiteur de ligne pour l'importation CSV. Valeur par défaut : \n. Si vos données contiennent \t ou \n, vous devez spécifier d'autres caractères comme délimiteurs. session_variable : un en-tête HTTP de StarRocks Stream Load qui définit les variables de session StarRocks lors du chargement des données. Séparez plusieurs variables par des virgules (,). Utilisez ce paramètre pour contrôler le comportement de la session StarRocks pendant le chargement, tel que le groupe de calcul et le délai d'expiration. `{"column_separator":"\\x01","row_delimiter":"\\x02","session_variable":"warehouse=qwe,query_timeout=600"}` Toutes les paires clé-valeur configurées dans `loadProps` sont transmises à l'interface StarRocks Stream Load en tant qu'en-têtes HTTP. Par conséquent, vous pouvez configurer n'importe quel paramètre d'en-tête pris en charge par StarRocks Stream Load, tel que `session_variable`, `max_filter_ratio` et `timeout`, dans `loadProps`. Stream Load prend également en charge l'importation de données JSON. Pour ce faire, configurez le paramètre suivant : `{ "format": "json" }` Les paramètres suivants sont disponibles pour le format JSON : strip_outer_array : indique s'il faut supprimer la structure de tableau la plus externe. Valeurs valides : `true` et `false`. Valeur par défaut : `false`. Les données JSON à importer peuvent être entourées de crochets externes (`[]`) représentant un tableau. Dans ce cas, définissez ce paramètre sur `true`. StarRocks supprime alors les crochets externes (`[]`) et importe chaque tableau interne en tant que ligne distincte. Si la valeur est définie sur `false`, StarRocks analyse l'intégralité du fichier de données JSON en tant que tableau unique et l'importe en une seule ligne. `[{"category":1,"author":2},{"category":3,"author":4}]` Si la valeur est définie sur `true`, StarRocks analyse `{"category":1,"author":2}` et `{"category":3,"author":4}` en deux lignes et les importe dans la table StarRocks de destination. Si la valeur est définie sur `false`, StarRocks analyse l'intégralité du tableau JSON en tant que ligne unique et l'importe dans la table StarRocks de destination. ignore_json_size : indique s'il faut vérifier la taille du corps JSON dans une requête HTTP. Remarque Par défaut, la taille du corps JSON dans une requête HTTP ne peut pas dépasser `100 Mo`. Si la taille du corps JSON dépasse `100 Mo`, le message d'erreur suivant est renvoyé : `The size of this batch exceed the max size [104857600] of json type data data [8617627793].Set ignore_json_size to skip check,although it may lead huge memory consuming.` Pour éviter cette erreur, vous pouvez ajouter `ignore_json_size: true` à l'en-tête de la requête `HTTP` pour ignorer la vérification de la taille du corps JSON. compression : spécifie l'algorithme de compression à utiliser lors de la transmission des données Stream Load. Algorithmes pris en charge : `GZIP`, `BZIP2`, `LZ4_FRAME` et `ZSTD`. strict_mode : indique s'il faut activer le mode strict. Valeurs valides : `true` : Active le mode strict. StarRocks filtre les lignes invalides, importe uniquement les lignes valides et renvoie des détails sur les données invalides. `false` : Désactive le mode strict. StarRocks convertit les champs qui échouent à la conversion en valeurs `NULL` et importe les lignes contenant ces valeurs `NULL` ainsi que les lignes de données valides. Valeur par défaut : `false`.	Oui	Aucun
column	Les colonnes de destination vers lesquelles vous souhaitez synchroniser les données.	Oui	Aucun
loadUrl	L'adresse IP et le port HTTP du nœud frontal (FE) StarRocks. Le port est par défaut `8030`. Pour plusieurs nœuds FE, spécifiez-les tous séparés par des virgules (,).	Oui	Aucun
table	La table de destination vers laquelle vous souhaitez synchroniser les données.	Oui	Aucun
preSql	Instructions SQL à exécuter avant le démarrage de la tâche de synchronisation. Par exemple, exécutez `TRUNCATE TABLE tablename` pour effacer les données existantes de la table de destination.	Non	Aucun
postSql	Instructions SQL à exécuter après la fin de la tâche de synchronisation.	Non	Aucun
maxBatchRows	Le nombre maximal de lignes à écrire par lot.	Non	500000
maxBatchSize	La taille maximale des données à écrire par lot, en octets.	Non	5242880
strategyOnError	La politique de gestion des erreurs survenant lors des écritures par lots. Valeurs valides : `exit` : Si une erreur d'écriture se produit, la tâche échoue et se termine. `batchDirtyData` : Si une erreur d'écriture se produit, la tâche enregistre le lot de données actuel en tant que données incorrectes. Valeur par défaut : `exit`.	Non	exit