Conector SQL do StarRocks - Realtime Compute for Apache Flink: Central de Documentação do Alibaba Cloud

Contexto

O StarRocks é um data warehouse MPP (Massively Parallel Processing) de nova geração que oferece desempenho extremamente rápido em todos os cenários e uma experiência analítica unificada. O StarRocks apresenta as seguintes vantagens:

É compatível com o protocolo MySQL, o que permite usar clientes MySQL e ferramentas comuns de Business Intelligence (BI) para conectar-se e analisar dados.
Adota arquitetura distribuída:
- Particiona horizontalmente as tabelas de dados e as armazena com múltiplas réplicas.
- Permite dimensionar o cluster flexivelmente e analisar até 10 petabytes (PB) de dados.
- Usa framework MPP para acelerar computações paralelas.
- Suporta múltiplas réplicas para garantir tolerância a falhas.

O conector Flink armazena dados em cache e usa o Stream Load para gravá-los em lotes nas tabelas de destino. Ele lê as tabelas de source buscando dados em lotes. A tabela a seguir lista as capacidades do conector StarRocks.

Categoria	Descrição
Tipos suportados	Tabelas de source, tabelas de dimensão, tabelas de destino e alvos de ingestão de dados
Modo de execução	Modo streaming e modo batch
Formato de dados	CSV
Métricas específicas do conector	Nenhuma
Tipos de API	DataStream, SQL e yaml para ingestão de dados
Suporte a atualizações/exclusões em tabelas de destino	Sim

Pré-requisitos

Você deve ter um cluster StarRocks implantado no EMR ou um cluster autogerenciado no ECS.

Limitações

Somente o Ververica Runtime (VVR) 11.1 ou superior suporta joins com tabelas de dimensão.
Para evitar restrições de acesso à rede, adicione as seguintes portas do cluster StarRocks à lista de permissões do grupo de segurança ou firewall: 9030, 8030, 8040, 9060, 8060, 9020.
Se a tabela StarRocks de destino contiver colunas geradas ocultas, declare apenas as colunas físicas reais a serem gravadas no job Flink e exclua os campos das colunas geradas. Colunas geradas ocultas criadas automaticamente pelo particionamento por expressão do StarRocks (como __generated_partition_column_0) não aceitam gravações externas. Por padrão, o conector constrói solicitações de escrita com base no schema completo, o que causa falha no job. Para mais informações sobre colunas geradas no StarRocks, consulte Generated columns.

Recursos

O StarRocks no E-MapReduce suporta as instruções CREATE TABLE AS SELECT (CTAS) e CREATE DATABASE AS SELECT (CDAS). O CTAS sincroniza o schema e os dados de uma única tabela, enquanto o CDAS sincroniza um banco de dados inteiro ou várias tabelas dentro do mesmo banco de dados. Para mais informações, consulte Use CTAS and CDAS statements in Realtime Compute for Apache Flink to synchronize data from a MySQL database to StarRocks.

Sintaxe

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	Descrição	Tipo	Obrigatório	Padrão	Observações
Geral	connector	Defina o conector a ser usado.	String	Sim	—	O valor deve ser `starrocks`.
	jdbc-url	URL do Java Database Connectivity (JDBC).	String	Sim	—	Especifique o endereço IP e a porta JDBC do FE no formato `jdbc:mysql://ip:port`.
	database-name	Nome do banco de dados StarRocks.	String	Sim	—	—
	table-name	Nome da tabela StarRocks.	String	Sim	—	—
	username	Nome de usuário para conexão ao StarRocks.	String	Sim	—	—
	password	Senha para conexão ao StarRocks.	String	Sim	—	—
	starrocks.create.table.properties	Define propriedades para criação automática de tabelas.	String	Não	—	Especifica propriedades iniciais da tabela, como engine e quantidade de réplicas. Exemplo: 'starrocks.create.table.properties' = 'buckets 8' ou 'starrocks.create.table.properties' = 'replication_num=1'.
Específico de source	scan-url	URL para varredura de dados.	String	Não	—	Especifica o endereço IP e a porta HTTP do FE. Formato: `fe_ip:http_port;fe_ip:http_port`. Nota Para especificar vários endereços IP e portas, separe-os por vírgulas (,).
	scan.connect.timeout-ms	Tempo limite para o `flink-connector-starrocks` se conectar ao StarRocks. O conector reporta um erro caso a conexão não seja estabelecida dentro desse prazo.	String	Não	1000	Unidade: milissegundos.
	scan.params.keep-alive-min	Duração keep-alive para a tarefa de consulta.	String	Não	10	—
	scan.params.query-timeout-s	Tempo limite para uma tarefa de consulta. Caso nenhum resultado seja retornado nesse período, o sistema interrompe a tarefa de consulta.	String	Não	600	Unidade: segundos.
	scan.params.mem-limit-byte	Limite de memória para uma única consulta em um nó BE.	String	Não	1073741824 (1 GB)	Unidade: bytes.
	scan.max-retries	Número máximo de tentativas para uma consulta com falha. O conector reporta um erro se esse limite for excedido.	String	Não	1	—
Específico de destino	load-url	URL para importação de dados.	String	Sim	—	Especifique os endereços IP e as portas HTTP dos FEs no formato `fe_ip:http_port;fe_ip:http_port`. Nota Para especificar vários endereços IP e portas, separe-os por ponto e vírgula (;).
	sink.semantic	Semântica de entrega para gravações.	String	Não	at-least-once	Valores válidos: `at-least-once` (padrão): Garante que os dados sejam entregues pelo menos uma vez. `exactly-once`: Garante que os dados sejam entregues exatamente uma vez.
	sink.buffer-flush.max-bytes	Quantidade máxima de dados a serem armazenados em buffer antes do flush.	String	Não	94371840 (90 MB)	Intervalo válido: 64 MB a 10 GB.
	sink.buffer-flush.max-rows	Número máximo de linhas a serem armazenadas em buffer antes do flush.	String	Não	500000	Intervalo válido: 1.000 a 5.000.000.
	sink.buffer-flush.interval-ms	Intervalo de flush do buffer.	String	Não	300000	Intervalo válido: 1.000 ms a 3.600.000 ms.
	sink.max-retries	Número máximo de tentativas para gravações com falha.	String	Não	3	Intervalo válido: 0 a 1000.
	sink.connect.timeout-ms	Tempo limite para conexão ao StarRocks.	String	Não	1000	Intervalo válido: 100 a 60.000. Unidade: milissegundos.
	sink.properties.*	Propriedades adicionais do Stream Load para o destino.	String	Não	—	Esses parâmetros controlam o comportamento do Stream Load. Por exemplo, `sink.properties.format` define o formato dos dados importados, como CSV. Para mais parâmetros, consulte Stream Load.
Específico de dimensão	lookup.cache.enabled	Defina se o cache deve ser ativado para a tabela de dimensão.	Boolean	Não	true	Valores válidos: `true`: Ative o cache. Após a primeira leitura dos dados da tabela, o sistema os armazena em cache na memória. Solicitações subsequentes usam os dados em cache dentro de seu período de validade para reduzir a sobrecarga de I/O. `false`: Desativa o cache. Cada consulta acessa diretamente a fonte de dados. Importante Este recurso requer o mecanismo Realtime Compute for Apache Flink VVR 11.1 ou superior. Recomendamos desativar este recurso nos seguintes cenários: Os dados da tabela de dimensão são atualizados frequentemente e dados em tempo real são necessários. A tabela contém uma grande quantidade de dados, apresentando risco de estouro de memória.

Mapeamento de tipos de dados

Tipo de dado StarRocks	Tipo de dado Flink
NULL	NULL
BOOLEAN	BOOLEAN
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
BIGINT UNSIGNED Nota Requer o mecanismo Realtime Compute for Apache Flink VVR 8.0.10 ou superior.	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 O VVR 8.0.10 estende automaticamente o comprimento CHAR em três vezes (m=n3, onde n<=85) para acomodar diferenças de codificação entre MySQL e StarRocks. O VVR 8.0.11 e versões posteriores estendem automaticamente o comprimento CHAR em quatro vezes (m=n4, onde n<=63) para acomodar diferenças de codificação entre MySQL e StarRocks. O comprimento máximo para o tipo CHAR do StarRocks é 255. Portanto, o Flink mapeia um tipo CHAR para um tipo CHAR do StarRocks somente se seu comprimento estendido automaticamente não exceder 255.	CHAR(n)
VARCHAR(m) Nota O VVR 8.0.10 estende automaticamente o comprimento VARCHAR em três vezes (m=n3, onde n>85) para acomodar diferenças de codificação entre MySQL e StarRocks. O VVR 8.0.11 e versões posteriores estendem automaticamente o comprimento VARCHAR em quatro vezes (m=n4, onde n>63) para acomodar diferenças de codificação entre MySQL e StarRocks. O comprimento máximo para o tipo CHAR do StarRocks é 255. Portanto, se o comprimento estendido automaticamente de um tipo CHAR do Flink exceder 255, o Flink mapeia o tipo para o tipo VARCHAR do StarRocks.	CHAR(n)
VARCHAR	STRING
VARBINARY Nota Requer o mecanismo Realtime Compute for Apache Flink VVR 8.0.10 ou superior.	VARBINARY

Exemplos

As tabelas de source, de destino e de dimensão nos exemplos a seguir podem ser usadas em conjunto. A tabela runoob_tbl_source declarada no exemplo de tabela de source também serve como fonte de dados para os exemplos de tabela de destino e de dimensão. Antes de usar estes exemplos, prepare tabelas com tipos de campo correspondentes no StarRocks e substitua os endpoints, nomes de bancos de dados e tabelas, além das variáveis secretas.

Tabela de 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

A tabela de source é lida em modo batch. Após o início do job, ele varre os dados atuais da tabela e então para de produzir saída. Ele não produz alterações incrementais continuamente.

Tabela de destino

Nota

O StarRocks permite que uma coluna de chave primária seja NULLABLE. No entanto, o Flink não suporta uma chave primária que contenha uma coluna anulável. O modelo de consistência de dados do Flink exige que uma chave primária seja única e não anulável. Caso contrário, o Flink lança o erro Invalid primary key. Column 'xxx' is nullable. Para mais informações, 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+

No VVR 8.x, o valor padrão de sink.version é V1, sendo suportados V1, V2 e AUTO. Parâmetros registrados apenas no VVR 11.x, como sink.ignore.update-before, não são suportados:

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;

Tabela de dimensão

Joins com tabelas de dimensão são suportados apenas no VVR 11.1 e versões posteriores. A tabela de dimensão deve declarar uma chave primária como chave de join, e o temporal join utiliza o tempo de processamento. O exemplo a seguir baseia-se na tabela runoob_tbl_source do exemplo de tabela de source, que inclui a coluna de tempo de processamento 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;