Conector SQL de StarRocks - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Aprenda a usar el conector de StarRocks en trabajos SQL.

Contexto

StarRocks es un almacén de datos Massively Parallel Processing (MPP) de nueva generación que ofrece un rendimiento extremadamente rápido en todos los escenarios y una experiencia de análisis unificada. StarRocks presenta las siguientes ventajas:

Es compatible con el protocolo MySQL, por lo que puede usar clientes MySQL y herramientas comunes de Business Intelligence (BI) para conectarse y analizar datos.
Utiliza una arquitectura distribuida:
- Particiona horizontalmente las tablas de datos y las almacena con varias réplicas.
- El clúster se escala de forma flexible y analiza hasta 10 petabytes (PB) de datos.
- Usa un framework MPP para acelerar los cálculos paralelos.
- Admite varias réplicas para proporcionar tolerancia a fallos.

El conector de Flink almacena los datos en caché y usa Stream Load para escribirlos por lotes en las tablas sink. Lee las tablas source al obtener los datos por lotes. A continuación se muestran las capacidades del conector de StarRocks.

Categoría	Descripción
Tipos compatibles	Tablas source, tablas de dimensiones, tablas sink y destinos de ingesta de datos
Modo de ejecución	Modo streaming y modo por lotes
Formato de datos	CSV
Métricas específicas del conector	Ninguna
Tipos de API	DataStream, SQL y YAML para la ingesta de datos
Compatibilidad con actualizaciones o eliminaciones en tablas sink	Sí

Requisitos previos

Debe tener un clúster de StarRocks desplegado en EMR o uno autogestionado en ECS.

Limitaciones

Solo Ververica Runtime (VVR) 11.1 o versiones posteriores admiten uniones con tablas de dimensiones.
Para evitar restricciones de acceso a la red, agregue los siguientes puertos del clúster de StarRocks a una lista de permisos de security group o firewall: 9030, 8030, 8040, 9060, 8060, 9020.
Cuando la tabla de destino de StarRocks contiene columnas generadas ocultas, declare únicamente las columnas físicas reales que se van a escribir en el trabajo de Flink y excluya los campos de la columna generada. Las columnas generadas ocultas creadas automáticamente por el particionamiento por expresiones de StarRocks (como __generated_partition_column_0) no aceptan escrituras externas. De forma predeterminada, el conector construye las solicitudes de escritura con base en el esquema completo, lo que provoca que el trabajo termine con errores. Para obtener más información sobre las columnas generadas en StarRocks, consulte Columnas generadas.

Funciones

StarRocks en E-MapReduce admite las instrucciones CREATE TABLE AS SELECT (CTAS) y CREATE DATABASE AS SELECT (CDAS). CTAS sincroniza el esquema y los datos de una sola tabla, mientras que CDAS sincroniza una base de datos completa o varias tablas dentro de la misma base de datos. Para obtener más información, consulte Use CTAS and CDAS statements in Realtime Compute for Apache Flink to synchronize data from a MySQL database to StarRocks.

Sintaxis

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

Parámetros

Tipo	Parámetro	Descripción	Tipo de dato	Obligatorio	Predeterminado	Observaciones
General	connector	Especifica el conector que se va a utilizar.	String	Sí	—	El valor debe ser `starrocks`.
	jdbc-url	La URL de Java Database Connectivity (JDBC).	String	Sí	—	Especifique la dirección IP y el puerto JDBC del FE con el formato `jdbc:mysql://ip:port`.
	database-name	El nombre de la base de datos de StarRocks.	String	Sí	—	—
	table-name	El nombre de la tabla de StarRocks.	String	Sí	—	—
	username	El nombre de usuario para conectarse a StarRocks.	String	Sí	—	—
	password	La contraseña para conectarse a StarRocks.	String	Sí	—	—
	starrocks.create.table.properties	Especifica las propiedades para la creación automática de tablas.	String	No	—	Define las propiedades iniciales de la tabla, como el motor y la cantidad de réplicas. Por ejemplo: 'starrocks.create.table.properties' = 'buckets 8' o 'starrocks.create.table.properties' = 'replication_num=1'.
Específicos de source	scan-url	La URL de escaneo de datos.	String	No	—	Especifica la dirección IP y el puerto HTTP del FE. Formato: `fe_ip:http_port;fe_ip:http_port`. Nota Para especificar varias direcciones IP y puertos, sepárelos con comas (,).
	scan.connect.timeout-ms	El tiempo de espera para que `flink-connector-starrocks` se conecte a StarRocks. El conector reporta un error si no se establece la conexión dentro de este tiempo de espera.	String	No	1000	Unidad: milisegundos.
	scan.params.keep-alive-min	La duración de keep-alive para la tarea de consulta.	String	No	10	—
	scan.params.query-timeout-s	El tiempo de espera para una tarea de consulta. Si no se devuelve ningún resultado durante este periodo, el sistema detiene la tarea de consulta.	String	No	600	Unidad: segundos.
	scan.params.mem-limit-byte	El límite de memoria para una sola consulta en un nodo BE.	String	No	1073741824 (1 GB)	Unidad: bytes.
	scan.max-retries	El número máximo de reintentos para una consulta con errores. El conector reporta un error si se supera este límite.	String	No	1	—
Específicos de sink	load-url	La URL de importación de datos.	String	Sí	—	Especifique las direcciones IP y los puertos HTTP de los FE con el formato `fe_ip:http_port;fe_ip:http_port`. Nota Para especificar varias direcciones IP y puertos, sepárelos con puntos y comas (;).
	sink.semantic	La semántica de entrega para las escrituras.	String	No	at-least-once	Valores válidos: `at-least-once` (predeterminado): garantiza que los datos se entreguen como mínimo una vez. `exactly-once`: garantiza que los datos se entreguen exactamente una vez.
	sink.buffer-flush.max-bytes	La cantidad máxima de datos que se almacenan en búfer antes de vaciarlo.	String	No	94371840 (90 MB)	Rango válido: de 64 MB a 10 GB.
	sink.buffer-flush.max-rows	El número máximo de filas que se almacenan en búfer antes de vaciarlo.	String	No	500000	Rango válido: de 1,000 a 5,000,000.
	sink.buffer-flush.interval-ms	El intervalo de vaciado del búfer.	String	No	300000	Rango válido: de 1,000 ms a 3,600,000 ms.
	sink.max-retries	El número máximo de reintentos para escrituras con errores.	String	No	3	Rango válido: de 0 a 1000.
	sink.connect.timeout-ms	El tiempo de espera para conectarse a StarRocks.	String	No	1000	Rango válido: de 100 a 60,000. Unidad: milisegundos.
	sink.properties.*	Propiedades adicionales de Stream Load para el destino.	String	No	—	Estos parámetros controlan el comportamiento de Stream Load. Por ejemplo, `sink.properties.format` define el formato de los datos importados, como CSV. Para consultar más parámetros, consulte Stream Load.
Específicos de dimensiones	lookup.cache.enabled	Indica si se habilita la caché para la tabla de dimensiones.	Boolean	No	true	Valores válidos: `true`: habilita la caché. Después de leer los datos de la tabla por primera vez, el sistema los almacena en memoria. Las solicitudes posteriores usan los datos almacenados en caché durante su periodo de validez para reducir la sobrecarga de E/S. `false`: deshabilita la caché. Cada consulta accede directamente al origen de datos. Importante Esta función requiere el motor Realtime Compute for Apache Flink VVR 11.1 o posterior. Se recomienda deshabilitar esta función en los siguientes escenarios: Los datos de la tabla de dimensiones se actualizan con frecuencia y se requieren datos en tiempo real. La tabla contiene una gran cantidad de datos, lo que representa un riesgo de desbordamiento de memoria.

Asignación de tipos de datos

Tipo de datos de StarRocks	Tipo de datos de Flink
NULL	NULL
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
BIGINT UNSIGNED Nota Requiere el motor Realtime Compute for Apache Flink VVR 8.0.10 o posterior.	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) Nota VVR 8.0.10 extiende automáticamente la longitud de CHAR al triple (m=n3, donde n<=85) para adaptarse a las diferencias de codificación entre MySQL y StarRocks. VVR 8.0.11 y versiones posteriores extienden automáticamente la longitud de CHAR al cuádruple (m=n4, donde n<=63) para adaptarse a las diferencias de codificación entre MySQL y StarRocks. La longitud máxima del tipo CHAR de StarRocks es 255. Por lo tanto, Flink asigna un tipo CHAR al tipo CHAR de StarRocks solo si su longitud extendida automáticamente no supera 255.	CHAR(n)
VARCHAR(m) Nota VVR 8.0.10 extiende automáticamente la longitud de VARCHAR al triple (m=n3, donde n>85) para adaptarse a las diferencias de codificación entre MySQL y StarRocks. VVR 8.0.11 y versiones posteriores extienden automáticamente la longitud de VARCHAR al cuádruple (m=n4, donde n>63) para adaptarse a las diferencias de codificación entre MySQL y StarRocks. La longitud máxima del tipo CHAR de StarRocks es 255. Por lo tanto, si la longitud extendida automáticamente de un tipo CHAR de Flink supera 255, Flink asigna ese tipo al tipo VARCHAR de StarRocks.	CHAR(n)
VARCHAR	STRING
VARBINARY Nota Requiere el motor Realtime Compute for Apache Flink VVR 8.0.10 o posterior.	VARBINARY

Ejemplos

La tabla source, la tabla sink y la tabla de dimensiones de los siguientes ejemplos se pueden usar en conjunto. La tabla runoob_tbl_source declarada en el ejemplo de tabla source también sirve como origen de datos para los ejemplos de tabla sink y de tabla de dimensiones. Antes de usar estos ejemplos, prepare tablas con tipos de campo coincidentes en StarRocks y reemplace los endpoints, los nombres de bases de datos y tablas, así como las variables secretas.

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

Nota

La tabla source se lee en modo por lotes. Una vez que inicia el trabajo, este escanea los datos actuales de la tabla y luego deja de generar resultados. No produce cambios incrementales de forma continua.

Tabla sink

Nota

StarRocks permite que una columna de clave primaria sea NULLABLE. Sin embargo, Flink no admite una clave primaria que contenga una columna anulable. El modelo de consistencia de datos de Flink exige que la clave primaria sea única y no anulable. De lo contrario, Flink genera el error Invalid primary key. Column 'xxx' is nullable. Para obtener más información, consulte "Invalid primary key. Column 'xxx' is nullable." error.

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+

En VVR 8.x, el valor predeterminado de sink.version es V1, y se admiten V1, V2 y AUTO. Los parámetros registrados únicamente en VVR 11.x, como sink.ignore.update-before, no son compatibles:

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;

Tabla de dimensiones

Las uniones con tablas de dimensiones solo se admiten en VVR 11.1 y versiones posteriores. La tabla de dimensiones debe declarar una clave primaria como clave de unión, y la unión temporal utiliza el tiempo de procesamiento. El siguiente ejemplo se basa en la tabla runoob_tbl_source del ejemplo de tabla source, la cual incluye la columna de tiempo de procesamiento 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;