Connecteur SQL StarRocks - Realtime Compute for Apache Flink

Contexte

StarRocks est un entrepôt de données MPP (Massively Parallel Processing) de nouvelle génération qui offre des performances extrêmement rapides dans tous les scénarios ainsi qu'une expérience analytique unifiée. StarRocks présente les avantages suivants :

StarRocks est compatible avec le protocole MySQL, ce qui vous permet d'utiliser des clients MySQL et des outils courants de Business Intelligence (BI) pour vous connecter aux données et les analyser.
StarRocks repose sur une architecture distribuée :
- Il partitionne horizontalement les tables de données et les stocke avec plusieurs réplicas.
- Le cluster peut être mis à l'échelle de manière flexible et peut analyser jusqu'à 10 pétaoctets (Po) de données.
- Il utilise un framework MPP pour accélérer les calculs parallèles.
- Il prend en charge plusieurs réplicas afin d'assurer la tolérance aux pannes.

Le connecteur Flink met les données en cache et utilise Stream Load pour les écrire par lots dans les tables de destination (sink). Il lit les tables source en récupérant les données par lots. Le tableau suivant répertorie les fonctionnalités du connecteur StarRocks.

Catégorie

Description

Types pris en charge

Tables source, tables de dimension, tables de destination et cibles d'ingestion de données

Mode d'exécution

Mode streaming et mode batch

Format de données

CSV

Métriques spécifiques au connecteur

Aucune

Types d'API

DataStream, SQL et YAML pour l'ingestion de données

Prise en charge des mises à jour/suppressions dans les tables de destination

Oui

Prérequis

Vous devez disposer d'un cluster StarRocks déployé sur EMR ou d'un cluster auto-géré sur ECS.

Limitations

Seules les versions Ververica Runtime (VVR) 11,1 et ultérieures prennent en charge les jointures avec les tables de dimension.
Pour éviter les restrictions d'accès réseau, ajoutez les ports suivants du cluster StarRocks à un groupe de sécurité ou à une liste d'autorisation de pare-feu : 9030, 8030, 8040, 9060, 8060, 9020.
Lorsque la table StarRocks cible contient des colonnes générées masquées, déclarez uniquement les colonnes physiques réelles à écrire dans le job Flink, et excluez les champs de colonne générée. Les colonnes générées masquées créées automatiquement par le partitionnement par expression StarRocks (telles que __generated_partition_column_0) n'acceptent pas les écritures externes. Par défaut, le connecteur construit les requêtes d'écriture en se basant sur le schéma complet, ce qui entraîne l'échec du job. Pour plus d'informations sur les colonnes générées dans StarRocks, consultez la documentation Colonnes générées.

Fonctionnalités

StarRocks sur E-MapReduce prend en charge les instructions CREATE TABLE AS SELECT (CTAS) et CREATE DATABASE AS SELECT (CDAS). CTAS synchronise le schéma et les données d'une seule table, tandis que CDAS synchronise une base de données entière ou plusieurs tables au sein de la même base de données. Pour plus d'informations, consultez Utiliser les instructions CTAS et CDAS dans Realtime Compute for Apache Flink pour synchroniser les données d'une base de données MySQL vers StarRocks.

Syntaxe

CREATE TABLE USER_RESULT(
 name VARCHAR,
 score BIGINT
 ) WITH (
 'connector' = 'starrocks',
 'jdbc-url'='jdbc:mysql://fe1_ip:query_port,fe2_ip:query_port,fe3_ip:query_port?xxxxx',
 'load-url'='fe1_ip:http_port;fe2_ip:http_port;fe3_ip:http_port',
 'database-name' = 'xxx',
 'table-name' = 'xxx',
 'username' = 'xxx',
 'password' = 'xxx'
 );

Paramètres

Type	Paramètre	Description	Type	Obligatoire	Valeur par défaut	Remarques
Général	connector	Spécifie le connecteur à utiliser.	String	Oui	—	La valeur doit être `starrocks`.
	jdbc-url	L'URL JDBC (Java Database Connectivity).	String	Oui	—	Spécifiez l'adresse IP et le port JDBC du FE au format `jdbc:mysql://ip:port`.
	database-name	Le nom de la base de données StarRocks.	String	Oui	—	—
	table-name	Le nom de la table StarRocks.	String	Oui	—	—
	username	Le nom d'utilisateur pour se connecter à StarRocks.	String	Oui	—	—
	password	Le mot de passe pour se connecter à StarRocks.	String	Oui	—	—
	starrocks.create.table.properties	Spécifie les propriétés pour la création automatique de table.	String	Non	—	Spécifie les propriétés initiales de la table, telles que le moteur et le nombre de réplicas. Par exemple : 'starrocks.create.table.properties' = 'buckets 8' ou 'starrocks.create.table.properties' = 'replication_num=1'.
Spécifique à la source	scan-url	L'URL de balayage des données.	String	Non	—	Spécifie l'adresse IP et le port HTTP du FE. Format : `fe_ip:http_port;fe_ip:http_port`. Remarque Pour spécifier plusieurs adresses IP et ports, séparez-les par des virgules (,).
	scan.connect.timeout-ms	Le délai d'expiration pour que `flink-connector-starrocks` se connecte à StarRocks. Le connecteur signale une erreur si une connexion n'est pas établie dans ce délai.	String	Non	1000	Unité : millisecondes.
	scan.params.keep-alive-min	La durée de maintien de la connexion (keep-alive) pour la tâche de requête.	String	Non	10	—
	scan.params.query-timeout-s	Le délai d'expiration pour une tâche de requête. Si aucun résultat n'est renvoyé dans ce délai, le système arrête la tâche de requête.	String	Non	600	Unité : secondes.
	scan.params.mem-limit-byte	La limite de mémoire pour une seule requête sur un nœud BE.	String	Non	1073741824 (1 Go)	Unité : octets.
	scan.max-retries	Le nombre maximal de tentatives pour une requête ayant échoué. Le connecteur signale une erreur si cette limite est dépassée.	String	Non	1	—
Spécifique à la destination (sink)	load-url	L'URL d'importation des données.	String	Oui	—	Spécifiez les adresses IP et les ports HTTP des FE au format `fe_ip:http_port;fe_ip:http_port`. Remarque Pour spécifier plusieurs adresses IP et ports, séparez-les par des points-virgules (;).
	sink.semantic	La sémantique de livraison pour les écritures.	String	Non	at-least-once	Valeurs valides : `at-least-once` (par défaut) : Garantit que les données sont livrées au moins une fois. `exactly-once` : Garantit que les données sont livrées exactement une fois.
	sink.buffer-flush.max-bytes	La quantité maximale de données à mettre en tampon avant vidage.	String	Non	94371840 (90 Mo)	Plage valide : 64 Mo à 10 Go.
	sink.buffer-flush.max-rows	Le nombre maximal de lignes à mettre en tampon avant vidage.	String	Non	500000	Plage valide : 1 000 à 5 000 000.
	sink.buffer-flush.interval-ms	L'intervalle de vidage du tampon.	String	Non	300000	Plage valide : 1 000 ms à 3 600 000 ms.
	sink.max-retries	Le nombre maximal de tentatives pour les écritures ayant échoué.	String	Non	3	Plage valide : 0 à 1000.
	sink.connect.timeout-ms	Le délai d'expiration pour la connexion à StarRocks.	String	Non	1000	Plage valide : 100 à 60 000. Unité : millisecondes.
	sink.properties.*	Propriétés Stream Load supplémentaires pour la destination.	String	Non	—	Ces paramètres contrôlent le comportement de Stream Load. Par exemple, `sink.properties.format` spécifie le format des données importées, tel que CSV. Pour plus de paramètres, consultez Stream Load.
Spécifique à la dimension	lookup.cache.enabled	Indique s'il faut activer la mise en cache pour la table de dimension.	Boolean	Non	true	Valeurs valides : `true` : Active la mise en cache. Après la première lecture des données de la table, le système les met en cache en mémoire. Les requêtes suivantes utilisent les données mises en cache pendant leur période de validité afin de réduire la charge d'E/S. `false` : Désactive la mise en cache. Chaque requête accède directement à la source de données. Important Cette fonctionnalité nécessite le moteur Realtime Compute for Apache Flink VVR 11.1 ou version ultérieure. Nous recommandons de désactiver cette fonctionnalité dans les scénarios suivants : Les données de la table de dimension sont fréquemment mises à jour et des données en temps réel sont requises. La table contient une grande quantité de données, ce qui pose un risque de dépassement de mémoire.

Mappage des types de données

Type de données StarRocks	Type de données Flink
NULL	NULL
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
BIGINT UNSIGNED Remarque Nécessite le moteur Realtime Compute for Apache Flink VVR 8.0.10 ou version ultérieure.	DECIMAL(20,0)
LARGEINT	DECIMAL(20,0)
FLOAT	FLOAT
DOUBLE	DOUBLE
DATE	DATE
DATETIME	TIMESTAMP
DECIMAL	DECIMAL
DECIMALV2	DECIMAL
DECIMAL32	DECIMAL
DECIMAL64	DECIMAL
DECIMAL128	DECIMAL
CHAR(m) Remarque VVR 8.0.10 étend automatiquement la longueur CHAR par trois (m=n3, où n<=85) pour tenir compte des différences d'encodage entre MySQL et StarRocks. VVR 8.0.11 et versions ultérieures étendent automatiquement la longueur CHAR par quatre (m=n4, où n<=63) pour tenir compte des différences d'encodage entre MySQL et StarRocks. La longueur maximale pour le type CHAR de StarRocks est de 255. Par conséquent, Flink mappe un type CHAR vers un type CHAR de StarRocks uniquement si sa longueur étendue automatiquement ne dépasse pas 255.	CHAR(n)
VARCHAR(m) Remarque VVR 8.0.10 étend automatiquement la longueur VARCHAR par trois (m=n3, où n>85) pour tenir compte des différences d'encodage entre MySQL et StarRocks. VVR 8.0.11 et versions ultérieures étendent automatiquement la longueur VARCHAR par quatre (m=n4, où n>63) pour tenir compte des différences d'encodage entre MySQL et StarRocks. La longueur maximale pour le type CHAR de StarRocks est de 255. Par conséquent, si la longueur étendue automatiquement d'un type CHAR Flink dépasse 255, Flink mappe le type vers le type VARCHAR de StarRocks.	CHAR(n)
VARCHAR	STRING
VARBINARY Remarque Nécessite le moteur Realtime Compute for Apache Flink VVR 8.0.10 ou version ultérieure.	VARBINARY

Exemples

La table source, la table de destination et la table de dimension des exemples suivants peuvent être utilisées conjointement. La table runoob_tbl_source déclarée dans l'exemple de table source sert également de source de données pour les exemples de table de destination et de table de dimension. Avant d'utiliser ces exemples, préparez des tables avec des types de champs correspondants dans StarRocks, et remplacez les endpoints, les noms de base de données et de table, ainsi que les variables secrètes.

Table source

CREATE TEMPORARY TABLE runoob_tbl_source (
    runoob_id BIGINT NOT NULL,
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    proc_time AS PROCTIME() --Processing time column, used by the temporal join in the dimension table example
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'scan-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with commas (,).
    'database-name' = '<database_name>',
    'table-name' = '<source_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'scan.params.query-timeout-s' = '600' --Timeout for a single query, in seconds
);

Remarque

La table source est lue en mode batch. Après le démarrage du job, il analyse les données actuelles de la table, puis cesse de produire une sortie. Il ne produit pas continuellement de modifications incrémentielles.

Table de destination

Remarque

StarRocks permet à une colonne de clé primaire d'être NULLABLE. Cependant, Flink ne prend pas en charge une clé primaire contenant une colonne nullable. Le modèle de cohérence des données de Flink exige qu'une clé primaire soit unique et non nullable. Sinon, Flink génère l'erreur Invalid primary key. Column 'xxx' is nullable. Pour plus d'informations, consultez Erreur « Invalid primary key. Column 'xxx' is nullable ».

VVR 11+

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, --Primary key columns must be declared as NOT NULL
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'load-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with semicolons (;).
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'sink.version' = 'V2', --Transactional Stream Load. Requires StarRocks 2.4 or later.
    'sink.semantic' = 'at-least-once', --Optional. exactly-once is also supported and requires checkpointing.
    'sink.buffer-flush.interval-ms' = '5000' --Batch flush interval, in milliseconds
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

VVR 8+

Dans VVR 8.x, la valeur par défaut de sink.version est V1, et les versions V1, V2 et AUTO sont prises en charge. Les paramètres enregistrés uniquement dans VVR 11.x, tels que sink.ignore.update-before, ne sont pas pris en charge :

CREATE TEMPORARY TABLE runoob_tbl_sink (
    runoob_id BIGINT NOT NULL, --Primary key columns must be declared as NOT NULL
    runoob_title STRING NOT NULL,
    runoob_author STRING NOT NULL,
    submission_date DATE,
    PRIMARY KEY (runoob_id) NOT ENFORCED
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'load-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with semicolons (;).
    'database-name' = '<database_name>',
    'table-name' = '<sink_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'sink.semantic' = 'at-least-once', --Optional. exactly-once is also supported and requires checkpointing.
    'sink.buffer-flush.interval-ms' = '5000', --Batch flush interval, in milliseconds
    'sink.max-retries' = '3' --Number of retries after a Stream Load failure
);

INSERT INTO runoob_tbl_sink
SELECT runoob_id, runoob_title, runoob_author, submission_date
FROM runoob_tbl_source;

Table de dimension

Les jointures de table de dimension sont prises en charge uniquement dans VVR 11.1 et versions ultérieures. La table de dimension doit déclarer une clé primaire comme clé de jointure, et la jointure temporelle utilise le temps de traitement. L'exemple suivant s'appuie sur la table runoob_tbl_source de l'exemple de table source, qui inclut la colonne de temps de traitement proc_time :

CREATE TEMPORARY TABLE sr_dim (
    runoob_id BIGINT NOT NULL,
    runoob_author STRING,
    PRIMARY KEY (runoob_id) NOT ENFORCED --The dimension table must declare a primary key as the join key
) WITH (
    'connector' = 'starrocks',
    'jdbc-url' = 'jdbc:mysql://<fe_host>:9030', --MySQL protocol endpoint of the FE. Separate multiple endpoints with commas (,).
    'scan-url' = '<fe_host>:8030', --HTTP endpoint of the FE. Separate multiple endpoints with commas (,).
    'database-name' = '<database_name>',
    'table-name' = '<dim_table_name>',
    'username' = '${secret_values.starrocks_username}', --Use secret variables to prevent credential leaks
    'password' = '${secret_values.starrocks_password}',
    'lookup.cache.ttl-ms' = '5000', --TTL of the dimension table cache, in milliseconds
    'lookup.cache.enabled' = 'true' --Default: true. Set to false to query the table directly without the cache.
);

SELECT o.runoob_id, o.runoob_title, d.runoob_author
FROM runoob_tbl_source AS o
JOIN sr_dim FOR SYSTEM_TIME AS OF o.proc_time AS d --Temporal join that uses the processing time column of the source table
ON o.runoob_id = d.runoob_id;