StarRocks 是一种高性能的分布式分析型数据库，广泛应用于实时数据分析场景。Apache Flink 作为流批一体化的计算框架，与 StarRocks 的结合能够高效处理大规模数据。为实现更高的读写性能，推荐使用 StarRocks 官方提供的 flink-connector-starrocks，而非 Flink 官方的 Flink JDBC Connector。

一、Flink-Connector-StarRocks 概述

1.1 为什么选择 Flink-Connector-StarRocks？

相较于 Flink JDBC Connector，StarRocks 自研的 Flink Connector 具有以下显著优势：

高性能并行读取：支持从 StarRocks 集群的各个 Backend（BE）节点并行读取数据，充分利用分布式架构，大幅提升读取效率。
高效批量写入：通过 Stream Load 机制，Flink 在内存中缓存小批量数据后以 HTTP PUT 请求的方式导入，StarRocks 的 Frontend（FE）将请求分发至 BE 节点，由 BE 节点作为 Coordinator 协调导入任务，降低 FE 负载。
事务支持：从 StarRocks 2.4 和 Flink Connector 1.2.4 开始，支持通过两阶段提交（2PC）实现 exactly-once 语义的写入。
灵活性与可扩展性：支持 DataStream API 和 Flink SQL 两种开发方式，满足不同开发需求。

1.2 适用场景

实时 ETL：从 StarRocks 读取数据，经过 Flink 加工后写入目标表。
数据分析：结合 Flink 的流批处理能力，对 StarRocks 数据进行实时或批量分析。
数据湖加速：通过高效读写，加速数据湖与 StarRocks 之间的数据交互。

二、Flink 读取 StarRocks 数据

Flink 支持通过 DataStream API 和 Flink SQL 从 StarRocks 表中读取数据。读取操作要求用户账号具备目标表的 SELECT 权限。

2.1 DataStream API 读取示例（基于 Flink 1.16）

以下是一个完整的 DataStream API 读取示例：

2.1.1 添加依赖

在项目中添加 flink-connector-starrocks 依赖：

<dependency><groupId>com.starrocks</groupId><artifactId>flink-connector-starrocks</artifactId><version>1.2.10_flink-1.16</version><!-- 对于 Flink 1.19，可使用 <version>1.2.10_flink-1.19</version> --><scope>provided</scope></dependency>

2.1.2 代码示例

public class StarRocksSourceApp {public static void main(String[] args) throws Exception {//构建 StarRocksSourceStarRocksSourceOptions options = StarRocksSourceOptions.builder()                          .withProperty("scan-url", "xxx:8030")                          .withProperty("jdbc-url", "jdbc:mysql://xxx:9030")                .withProperty("username", "xxx")                     .withProperty("password", "123456")             .withProperty("table-name", "score_board")            .withProperty("database-name", "unload")               .build();TableSchema tableSchema = TableSchema.builder()                 .field("id", DataTypes.INT())                          .field("name", DataTypes.STRING())                          .field("score", DataTypes.INT())                          .build();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.addSource(StarRocksSource.source(tableSchema, options)).setParallelism(5).print();env.execute("StarRocks flink source");    }}

2.2 Flink SQL 读取示例

Flink SQL 提供了更简洁的声明式方式，适合快速开发。

CREATE TABLE flink_sr_table(    
`id` INT,    
`name` STRING,    
`score` INT)
WITH(    
'connector'='starrocks',    
'scan-url'='xxx:8030',    
'jdbc-url'='jdbc:mysql://xxx:9030',    
'username'='xxx',    
'password'='xxx',    
'database-name'='unload',    
'table-name'='score_board');



select * from flink_sr_table;

三、Flink 写入 StarRocks 数据

Flink 支持通过 DataStream API 和 Flink SQL 向 StarRocks 表写入数据。写入操作要求用户账号具备目标表的 SELECT 和 INSERT 权限。

3.1 DataStream API 写入示例（基于 Flink 1.16）

3.1.1 添加依赖

与读取相同，添加以下依赖：

<dependency><groupId>com.starrocks</groupId><artifactId>flink-connector-starrocks</artifactId><version>1.2.10_flink-1.16</version><scope>provided</scope></dependency>

3.1.2 代码示例

public class StarRocksSinkApp {public static void main(String[] args) throws Exception {     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();     String[] records = new String[]{"500\tstarrocks-csv\t100","501\tflink-csv\t100"};     DataStream<String> source = env.fromElements(records);     StarRocksSinkOptions options = StarRocksSinkOptions.builder()                     .withProperty("jdbc-url", "jdbc:mysql://xxx:9030")                               .withProperty("load-url", "xxx:8030")                               .withProperty("database-name", "unload")                               .withProperty("table-name", "score_board")                               .withProperty("username", "xxx")                               .withProperty("password", "xxx")                               .withProperty("sink.properties.format", "csv")                               .withProperty("sink.properties.column_separator", "\t")                               .build();     // Create the sink with the options.     SinkFunction<String> starRockSink = StarRocksSink.sink(options);     source.addSink(starRockSink);     env.execute("StarRocks flink source");     }}

3.2 Flink SQL 写入示例

3.2.1 创建表

CREATE TABLE `score_board` (    
`id` INT,    
`name` STRING,    
`score` INT,    
PRIMARY KEY (id) NOT ENFORCED) 
WITH (   
'connector' = 'starrocks',    
'jdbc-url' = 'jdbc:mysql://xxx:9030',    
'load-url' = 'xxx:8030',    
'database-name' = 'unload',    
'table-name' = 'score_board',    
'username' = 'xxx',    
'password' = 'xxx');



INSERT INTO `score_board` VALUES (1, 'starrocks', 99), (2, 'flink', 100);

3.3 写入并发与批量控制

并发控制：

DataStream API：通过 Flink 作业的并行度设置（如 setParallelism）。
Flink SQL：通过 sink.parallelism 参数设置写入并发。

批量写入触发条件：

缓存数据大小达到 sink.buffer-flush.max-bytes。
缓存行数达到 sink.buffer-flush.max-rows。
距离上次刷新超过 sink.buffer-flush.interval-ms。

四、高性能实现原理

4.1 读取高性能实现

初始化：Flink 的 Source 任务通过 StarRocksSourceFunction 连接 StarRocks 的 FE，获取查询计划（QueryPlan），包含数据分片（Tablet）的分布信息。
并行读取：根据查询计划，Source 任务并行连接多个 BE 节点，每个子任务通过 TStarrocksExternalService.Client 使用 Thrift RPC 协议直接读取 Tablet 数据。
核心类：

StarRocksDynamicSourceFunction ：负责连接 FE、获取查询计划并分发读取任务。
StarRocksSourceBeReader ：通过 Thrift RPC 实现 BE 节点的数据读取。

在StarRocksDynamicSourceFunction构造函数中，会根据查询sql连接FE获取查询计划

open 方法，根据查询计划获取BE Tablet分区，同时构建对应的读取器StarRocksSrouceBeReader

run 方法，通过StarRocksSourceBeReader进行循环读取

StarRocksSourceBeReader里通过 Thrift RPC 对 BE 进行实际的数据读取

4.2 写入高性能实现

批量写入：通过 Stream Load（或事务性 Stream Load）将数据缓存到内存后批量发送，减少网络交互。
事务性 Stream Load（V2 模式）：

从 StarRocks 2.4 和 Flink 连接器 1.2.4 开始，支持事务性写入，通过两阶段提交（2PC）实现 exactly-once。

异步写入与重试：异步提交数据，支持配置重试次数（如 sink.max-retries ）。
内存管理：通过 sink.buffer-flush.max-rows 和 sink.buffer-flush.max-bytes 控制缓存数据量，优化吞吐量与延迟。

StarRocksDynamicSinkFunction，在invoke方法里对数据进行序列化，然后通过 StarRocksSinkManager 写入

DefaultStreamLoader，实现 Stream Load 的 HTTP 请求和响应处理，通过构建 HttpPut 生成对应的 Stream Load 任务

具体管理事务的生命周期 TransactionStreamLoader（begin、prepare、commit、 rollback）

StarRocksSinkManager，在内存攒一批数据，超过条件就flush

以上就是简单扼要的剖析flink-connector-starrocks对starrocks高效读写的原理，而jdbc本质还是通过mysql协议对starrocks进行读写操作，就必须经过FE，而这种方式，在读取比较大的结果集时，会对FE造成比较大的负担，因为结果集流量要通过FE；在写入的时候如果频繁的通过构造insert语句请求，也会对FE造成负担。

更多大数据干货，欢迎关注我的微信公众号—BigData共享

ITPub博客

Flink高性能读写StarRocks内表剖析

最新文章