一、application.yml

（因采用dynamic-datasource-spring-boot-starter動(dòng)態(tài)數(shù)據(jù)源，所以才是以下配置文件的樣式，像redis，druid根據(jù)自己情況導(dǎo)入依賴）

這個(gè)配置文件的場(chǎng)景是把starrocks當(dāng)成slave庫(kù)在用。某些大數(shù)據(jù)慢查詢就走starrocks

就這樣配置好后就可把starrocks當(dāng)mysql用了

# spring配置
spring:
  redis:
    host: localhost
    port: 6379
    password:
  datasource:
    druid:
      stat-view-servlet:
        enabled: true
        loginUsername: admin
        loginPassword: 123456
    dynamic:
      druid:
        initial-size: 5
        min-idle: 5
        maxActive: 20
        maxWait: 60000
        timeBetweenEvictionRunsMillis: 60000
        minEvictableIdleTimeMillis: 300000
        validationQuery: SELECT 1
        testWhileIdle: true
        testOnBorrow: false
        testOnReturn: false
        poolPreparedStatements: true
        maxPoolPreparedStatementPerConnectionSize: 20
        filters: stat,slf4j
        connectionProperties: druid.stat.mergeSql\=true;druid.stat.slowSqlMillis\=5000
      datasource:
        # 主庫(kù)數(shù)據(jù)源
        master:
          driver-class-name: com.mysql.cj.jdbc.Driver
          url: jdbc:mysql://xx.xx.xx.xx:3306/inst_ops?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&rewriteBatchedStatements=true
          username: root
          password: 123456
          # 從庫(kù)數(shù)據(jù)源
        slave:
          driver-class-name: com.mysql.cj.jdbc.Driver
          username: root
          password: 
          url: jdbc:mysql://xx.xx.xx.xx:9030/inst_ops?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=false&serverTimezone=GMT%2B8&rewriteBatchedStatements=true&autoReconnect=true&failOverReadOnly=false
      seata: true

# mybatis配置
# mybatis:
#     # 搜索指定包別名
#     typeAliasesPackage: com.ruoyi.stream
#     # 配置mapper的掃描，找到所有的mapper.xml映射文件
#     mapperLocations: classpath:mapper/**/*.xml

# swagger配置
swagger:
  title: mongodb接口文檔
  license: Powered By ruoyi
  licenseUrl: http://localhost:8080/doc.html

二、實(shí)驗(yàn)場(chǎng)景：查詢mysql數(shù)據(jù)，使用insert into批量往starrocks插入數(shù)據(jù)

重點(diǎn)：采用這種方式有限制，插入幾千條starrocks就不讓你插入了，只用于demo測(cè)試

package com.ruoyi.starrocks.test;

import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.ruoyi.starrocks.domain.DevMc;
import com.ruoyi.starrocks.mapper.DevMcMapper;
import com.ruoyi.starrocks.service.IDevMcService;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import javax.annotation.Resource;
import java.util.List;

/**
 *
 * @author yh
 * @date 2023-03-24 10:50
 */
@SpringBootTest
@Slf4j
public class StarrocksTest {

    public static final String NAME = "INSERT INTO `inst_ops`.dev_mc (id,sign_id,`year`,oid,ono,prov,city,area,park,tmpl," +
            "dtm,over_dtm,over_state,mc_tag,mc_cid,mc_uid,mc_man,mc_day,mc_sta,mc_end,audt_cid,audt_tag_chk,audt_tag_ok," +
            "audt_dtm,audt_uid,audt_man,audt_inf,submit,version,c_id,to_upd) VALUES " ;

    @Resource
    private IDevMcService devMcService;
    @Resource
    private DevMcMapper mapper;

    @Test
    public void test1() {
        LambdaQueryWrapper<DevMc> last = Wrappers.lambdaQuery(DevMc.class);
        List<DevMc> list = devMcService.list(last);
        System.err.println("查詢完畢");
        DynamicDataSourceContextHolder.push("slave"); // 手動(dòng)切換
        System.err.println("切換數(shù)據(jù)源成功---------------");
        StringBuilder sb = new StringBuilder();
        int count = 0;
        for (DevMc devMc : list) {
            count ++;
            sb.append("(");
            sb.append(devMc.getId()).append(",");
            sb.append(devMc.getSignId()).append(",");
            sb.append(devMc.getYear()).append(",");
            sb.append(devMc.getOid()).append(",");
            sb.append("'").append(devMc.getOno()).append("',");
            sb.append(devMc.getProv()).append(",");
            sb.append(devMc.getCity()).append(",");
            sb.append(devMc.getArea()).append(",");
            sb.append(devMc.getPark()).append(",");
            sb.append(devMc.getTmpl()).append(",");
            sb.append("'").append(devMc.getDtm()).append("',");
            sb.append("'").append(devMc.getOverDtm()).append("',");
            sb.append(devMc.getOverState()).append(",");
            sb.append(devMc.getMcTag()).append(",");
            sb.append(devMc.getMcCid()).append(",");
            sb.append("'").append(devMc.getMcUid()).append("',");
            sb.append("'").append(devMc.getMcMan()).append("',");
            sb.append("'").append(devMc.getMcDay()).append("',");
            sb.append("'").append(devMc.getMcSta()).append("',");
            sb.append("'").append(devMc.getMcEnd()).append("',");
            sb.append(devMc.getAudtCid()).append(",");
            sb.append(devMc.getAudtTagChk()).append(",");
            sb.append(devMc.getAudtTagOk()).append(",");
            sb.append("'").append(devMc.getAudtDtm()).append("',");
            sb.append(devMc.getAudtUid()).append(",");
            sb.append("'").append(devMc.getAudtMan()).append("',");
            sb.append("'").append(devMc.getAudtInf()).append("',");
            sb.append(devMc.getSubmit()).append(",");
            sb.append(devMc.getVersion()).append(",");
            sb.append(devMc.getCId()).append(",");
            sb.append(devMc.getToUpd());
            sb.append(")");
            if (count == 5000) {
                sb.append(";");
                StringBuilder insert = new StringBuilder(NAME);
                insert.append(sb.toString());
                mapper.insertBatch(insert.toString());
                count = 0;
                sb.setLength(0);
            } else {
                sb.append(",");
            }
        }

        if (sb.length() > 0) {
            StringBuilder insert = new StringBuilder(NAME);
            insert.append(sb.toString().substring(0, sb.length()-1));
            insert.append(";");
            mapper.insertBatch(insert.toString());
            count = 0;
            sb.setLength(0);
        }

    }

    @Test
    public void test2() {
        DevMc devMc = devMcService.getById(1558);
        System.err.println("查詢完畢");
        DynamicDataSourceContextHolder.push("slave"); // 手動(dòng)切換
        System.err.println("切換數(shù)據(jù)源成功---------------");
        devMcService.save(devMc);
    }
}

三、使用flink往starrocks導(dǎo)入數(shù)據(jù)，并實(shí)現(xiàn)亞秒級(jí)同步

第3.4章：StarRocks數(shù)據(jù)導(dǎo)入--Flink Connector與CDC秒級(jí)數(shù)據(jù)同步_flink-connector-starrocks_流木隨風(fēng)的博客-CSDN博客

所需文件：注意！有些jar包在外網(wǎng)沒(méi)得，只有阿里云倉(cāng)庫(kù)有。而且搜索的出來(lái)的jar包與博客展示的名稱有細(xì)微差別。按博客的名稱搜根本搜不出來(lái)，所以我覺得是他文檔好久沒(méi)更新了

flink版本：

flink-1.13.5-bin-scala_2.11.tgz、

需要jar包：

flink-connector-starrocks-1.1.14_flink-1.13_2.11.jar

flink-sql-connector-mysql-cdc-2.0.2.jar

flink-connector-jdbc_2.11-1.13.5.jar

jar包阿里網(wǎng)盤下載連接：阿里云盤分享

步驟就不用說(shuō)了，按那個(gè)博客進(jìn)行操作

四、相關(guān)建表語(yǔ)句(私人保留記錄用，跟文章無(wú)關(guān))

1、starrocks建庫(kù)語(yǔ)句

create database inst_ops;

2、starrocks建表語(yǔ)句

CREATE TABLE `dev_mc`  (
  `id` bigint NOT NULL COMMENT '維保流水號(hào)',
  `sign_id` bigint NOT NULL  COMMENT '簽到ID',
  `year` int NOT NULL  COMMENT '維保年份',
  `oid` int NOT NULL  COMMENT '設(shè)備編號(hào)',
  `ono` varchar(90)  NOT NULL DEFAULT '' COMMENT '設(shè)備號(hào)',
  `prov` int NOT NULL  COMMENT '省份',
  `city` int NOT NULL  COMMENT '城市',
  `area` int NOT NULL,
  `park` int NOT NULL,
  `tmpl` int NOT NULL COMMENT '模板編號(hào)',
  `dtm` varchar(57)  NOT NULL DEFAULT '' COMMENT '時(shí)間-維保最遲時(shí)間',
  `over_dtm` varchar(57)  NOT NULL DEFAULT '' COMMENT '維保超期時(shí)間',
  `over_state` tinyint(1) NOT NULL  COMMENT '超期是否計(jì)算',
  `mc_tag` tinyint(1) NOT NULL  COMMENT '是否維保',
  `mc_cid` int NOT NULL  COMMENT '維保單位',
  `mc_uid` varchar(90)  NOT NULL DEFAULT '0' COMMENT '維保人編號(hào)',
  `mc_man` varchar(90)  NOT NULL DEFAULT '' COMMENT '維保人姓名',
  `mc_day` varchar(33)  NOT NULL DEFAULT '' COMMENT '保養(yǎng)日期(xxxx年xx月xx日)',
  `mc_sta` varchar(57)  NOT NULL DEFAULT '' COMMENT '維保開始時(shí)間(xx時(shí)xx分)',
  `mc_end` varchar(57)  NOT NULL DEFAULT '' COMMENT '維保結(jié)束時(shí)間(xx時(shí)xx分)',
  `audt_cid` int NOT NULL  COMMENT '審核公司ID',
  `audt_tag_chk` tinyint(1) NOT NULL  COMMENT '使用單位是否審核',
  `audt_tag_ok` tinyint(1) NOT NULL  COMMENT '使用單位審核結(jié)果：通過(guò)/不通過(guò)',
  `audt_dtm` varchar(57)  NOT NULL DEFAULT '' COMMENT '審核日期',
  `audt_uid` int NOT NULL  COMMENT '使用單位審核人',
  `audt_man` varchar(90)  NOT NULL  COMMENT '使用單位審核人',
  `audt_inf` varchar(150)  NOT NULL  COMMENT '使用單位審核結(jié)果',
  `submit` tinyint(1) NOT NULL  COMMENT '是否提交物業(yè)審核',
  `version` int NOT NULL  COMMENT '版本號(hào)(1判斷審核后是否更新)',
  `c_id` bigint NOT NULL  COMMENT '96333返回ID',
  `to_upd` tinyint NULL COMMENT '是否要同步(0否1是)'
) PRIMARY KEY (id)
DISTRIBUTED BY HASH(id) BUCKETS 4
PROPERTIES("replication_num" = "1",
"enable_persistent_index" = "true");

3、flink?sql——mysql、starrocks的建表映射語(yǔ)句

mysql：

CREATE TABLE source_mysql_devmc (
  id BIGINT,
  sign_id BIGINT,
  `year` int ,
  `oid` int ,
  `ono` STRING  ,
  `prov` int ,
  `city` int,
  `area` int,
  `park` int ,
  `tmpl` int ,
  `dtm` STRING  ,
  `over_dtm` STRING  ,
  `over_state` BOOLEAN ,
  `mc_tag` BOOLEAN ,
  `mc_cid` int ,
  `mc_uid` STRING  ,
  `mc_man` STRING ,
  `mc_day` STRING ,
  `mc_sta` STRING  ,
  `mc_end` STRING ,
  `audt_cid` int ,
  `audt_tag_chk` BOOLEAN ,
  `audt_tag_ok` BOOLEAN ,
  `audt_dtm` STRING  ,
  `audt_uid` int ,
  `audt_man` STRING  ,
  `audt_inf` STRING , 
  `submit` BOOLEAN ,
  `version` int ,
  `c_id` bigint ,
  `to_upd` tinyint ,
   PRIMARY KEY (id) NOT ENFORCED
)WITH (
   'connector' = 'jdbc',
   'url' = 'jdbc:mysql://localhost:3306/inst_ops',
   'table-name' = 'dev_mc',
   'username' = 'root',
   'password' = '123456'
);

starrocks：

CREATE TABLE sink_starrocks_devmc (
  id BIGINT,
  sign_id BIGINT,
  `year` int ,
  `oid` int ,
  `ono` STRING  ,
  `prov` int ,
  `city` int,
  `area` int,
  `park` int ,
  `tmpl` int ,
  `dtm` STRING  ,
  `over_dtm` STRING  ,
  `over_state` BOOLEAN ,
  `mc_tag` BOOLEAN ,
  `mc_cid` int ,
  `mc_uid` STRING  ,
  `mc_man` STRING ,
  `mc_day` STRING ,
  `mc_sta` STRING  ,
  `mc_end` STRING ,
  `audt_cid` int ,
  `audt_tag_chk` BOOLEAN ,
  `audt_tag_ok` BOOLEAN ,
  `audt_dtm` STRING  ,
  `audt_uid` int ,
  `audt_man` STRING  ,
  `audt_inf` STRING , 
  `submit` BOOLEAN ,
  `version` int ,
  `c_id` bigint ,
  `to_upd` tinyint ,
   PRIMARY KEY (id) NOT ENFORCED
)WITH (
   'connector' = 'starrocks',
   'jdbc-url'='jdbc:mysql://localhost:9030',
   'load-url'='localhost:8030',
   'database-name' = 'inst_ops',
   'table-name' = 'dev_mc',
   'username' = 'root',
   'password' = '',
   'sink.buffer-flush.interval-ms' = '5000',
   'sink.properties.column_separator' = '\x01',
   'sink.properties.row_delimiter' = '\x02'
);

需要注意的點(diǎn)：因?yàn)槲襪ysql與starrocks都是用docker啟動(dòng)的，所以ip是localhost。用公網(wǎng)ip就會(huì)連不上數(shù)據(jù)庫(kù)，這是因?yàn)閐ocker自身的問(wèn)題，他們說(shuō)用紅帽版的docker就沒(méi)有這個(gè)問(wèn)題

4、同步數(shù)據(jù)語(yǔ)句

insert into sink_starrocks_devmc select * from source_mysql_devmc limit 300000;

5、亞秒級(jí)同步（只需要修改mysql映射）文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-810015.html

CREATE TABLE source_mysql_devmc (
  id BIGINT,
  sign_id BIGINT,
  `year` int ,
  `oid` int ,
  `ono` STRING  ,
  `prov` int ,
  `city` int,
  `area` int,
  `park` int ,
  `tmpl` int ,
  `dtm` STRING  ,
  `over_dtm` STRING  ,
  `over_state` BOOLEAN ,
  `mc_tag` BOOLEAN ,
  `mc_cid` int ,
  `mc_uid` STRING  ,
  `mc_man` STRING ,
  `mc_day` STRING ,
  `mc_sta` STRING  ,
  `mc_end` STRING ,
  `audt_cid` int ,
  `audt_tag_chk` BOOLEAN ,
  `audt_tag_ok` BOOLEAN ,
  `audt_dtm` STRING  ,
  `audt_uid` int ,
  `audt_man` STRING  ,
  `audt_inf` STRING , 
  `submit` BOOLEAN ,
  `version` int ,
  `c_id` bigint ,
  `to_upd` tinyint ,
   PRIMARY KEY (id) NOT ENFORCED
)WITH (
   'connector' = 'mysql-cdc',
   'hostname' = 'localhost',
   'port' = '3306',
   'username' = 'root',
   'password' = '123456',
   'database-name' = 'inst_ops',
   'scan.incremental.snapshot.enabled'='false',
   'table-name' = 'dev_mc'
);

到了這里，關(guān)于springboot集成starrocks、以及采用flink實(shí)現(xiàn)mysql與starrocks亞秒級(jí)同步的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！