流计算 Oceanus 数据仓库 StarRocks

一、版本说明

二、作为数据目的（Sink）

flink-connector-starrocks 作为数据目的，用于导入数据至 StarRocks，相比于 Flink 提供的 flink-connector-jdbc，导入性能更佳，flink-connector-starrocks 的内部实现是通过缓存并批量由 Stream Load 导入。支持 CSV、JSON 两种数据格式。

以下为 MySQL-CDC 数据实时导入 StarRocks 的示例。

CREATE TABLE `mysql_cdc` ( `user_id` bigint, `item_id` bigint, `behavior` STRING, PRIMARY KEY (`user_id`) NOT ENFORCED) WITH ( 'connector' = 'mysql-cdc', — 固定值 'mysql-cdc' 'hostname' = '9.134.34.15', — 数据库的 IP 'port' = '3306', — 数据库的访问端口 'username' = 'root', — 数据库访问的用户名（需要提供 SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT、SELECT 和 RELOAD 权限） 'password' = 'xxx', — 数据库访问的密码 'database-name' = 'test', — 需要同步的数据库 'table-name' = 'user_behavior' — 需要同步的数据表名);CREATE TABLE `pk_starrocks`( `user_id` bigint, `item_id` bigint, `behavior` STRING, PRIMARY KEY (`user_id`) NOT ENFORCED) WITH ( 'connector' = 'starrocks', 'jdbc-url' = 'jdbc:mysql://172.28.28.98:9030', 'load-url' = '172.28.28.98:8030', 'database-name' = 'oceanus', 'table-name' = 'pk_user_behavior', 'username' = 'root', 'password' = 'xxx', 'sink.buffer-flush.interval-ms' = '15000', 'sink.properties.format' = 'json', 'sink.properties.strip_outer_array' = 'true', — 需要设置为 true — 'sink.parallelism' = '1', 'sink.max-retries' = '3', 'sink.semantic' = 'exactly-once');INSERT INTO `pk_starrocks` SELECT * FROM `mysql_cdc`;

注意：说明：

三、WITH 参数

注意：说明：

四、类型映射（Sink）

五、StarRocks 数据模型

StarRocks 支持四种 数据模型：明细模型、聚合模型、更新模型、主键模型。主键模型支持谓词和索引下推，能够在支持实时和频繁更新等场景的同时，提供高效查询。如果没有特殊需求，推荐使用主键模型。

·对于 Upsert 流，需要使用主键模型，否则 DELETE 消息无法写入 StarRocks。

·相对于 Merge-On-Read 策略的更新模型，主键模型的查询性能能够提升 3~10 倍。

·主键模型可利用部分列更新实现多流 JOIN。

六、WITH 参数

七、批数据源

CREATE TABLE `starrocks` ( `user_id` bigint, `item_id` bigint, `behavior` STRING, PRIMARY KEY (`user_id`) NOT ENFORCED) WITH ( 'connector' = 'starrocks' , 'jdbc-url' = 'jdbc:mysql://172.28.28.98:9030', — query_port, FE mysql server port 'scan-url' = '172.28.28.98:8030', — http_port 'database-name' = 'oceanus', 'table-name' = 'pk_user_behavior', 'username' = 'root', 'password' = 'xxx');CREATE TABLE `print_sink` ( `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING, PRIMARY KEY (`user_id`) NOT ENFORCED) WITH ( 'connector' = 'logger');INSERT INTO `print_sink`SELECT * FROM starrocks;

八、维表

社区本身支持 starrocks connector 作为维表查询，但仍然推荐使用 JDBC connector，更加稳定和通用：

CREATE TABLE `starrocks` ( `user_id` bigint, `item_id` bigint, `behavior` STRING, PRIMARY KEY (`user_id`) NOT ENFORCED) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://172.28.28.98:9030/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai', — query_port, FE mysql server port 'table-name' = 'my-table', — 需要写入的数据表 'username' = 'admin', — 数据库访问的用户名（需要提供 INSERT 权限） 'password' = 'MyPa$$w0rd', — 数据库访问的密码 'lookup.cache.max-rows' = '100', — 读缓存大小 'lookup.cache.ttl' = '5000' — 读缓存的 TTL);CREATE TABLE `datagen` ( `user_id` BIGINT, `ts` as PROCTIME(), PRIMARY KEY (`user_id`) NOT ENFORCED) WITH ( 'connector' = 'datagen', 'rows-per-second' = '1', 'fields.user_id.min' = '1', 'fields.user_id.max' = '20');CREATE TABLE `print_sink` ( `user_id` BIGINT, `item_id` BIGINT, `behavior` STRING, `ts` TIMESTAMP, PRIMARY KEY (`user_id`) NOT ENFORCED) WITH ( 'connector' = 'logger');INSERT INTO `print_sink`SELECT a.user_id,b.item_id,b.behavior,a.tsFROM `datagen` a LEFT JOIN `starrocks` FOR SYSTEM_TIME AS OF a.ts as bON a.user_id = b.user_id;

九、类型映射（source）

十、更多常见问题

请参见 StarRocks Flink Connector 页面文档。