如何实时同步数据到StarRocks

我们知道,是StarRocks基于Doris开发的,它在多表连接查询的性能方面引领OLAP市场,是一个很好用的结构化数据仓库。但是一直没有很好的工具能够实现业务数据库的数据实时同步到StarRocks分布式数据仓库集群中,本文将带领读者一起来通过flink-jobs的最佳实践来构建一个易于使用和运维的,异构数据库实时同步到StarRocks集群的方案。

阅读前提

阅读本文除了需具备StarRocks相关知识之外,还会涉及Flink、Debezium、Kafka、Zookeeper、MySQL(或者PostgreSQL、Oracle、SQL Server、Db2等其中一种),最好能够自己搭建整套环境,如果不能,身边有可以利用的环境或者有朋友能帮上忙也可以。

环境准备

  1. 在192.168.100.1-3上安装zookeeper的,在192.168.100.4-6上安装基于zookeeper的Flink高可用集群,具体安装配置方法请查阅Flink官网
  2. 在192.168.100.7-9上安装基于zookeeper的kafka集群,并配置好Debezium(也可以是Canal、Maxwell等Flink支持的其他组件),安装方法详见各自官网。
  3. 在192.168.100.10上安装一个MySQL数据库,也可以安装任意Debezium支持的数据库(例如PostgreSQL、Oracle、SQL Server、Db2等)。创建一个skyline数据库,在skyline中创建product表并插入数据:
    -- ----------------------------
    -- Table structure for product
    -- ----------------------------
    CREATE TABLE product (
      id bigint NOT NULL,
      name VARCHAR(255),
      authors VARCHAR(50),
      create_time DATETIME,
      PRIMARY KEY (id)
    );
    
    -- ----------------------------
    -- Records of product
    -- ----------------------------
    INSERT INTO product VALUES (100000000000000001, 'Apache Flink', 'Apache Software Foundation', now());
    INSERT INTO product VALUES (100000000000000002, 'Flink Jobs', 'June', now());
  4. 192.168.100.11-13上安装一个StarRocks集群,创建一个skyline数据库,并在skyline中创建product表:
    -- ----------------------------
    -- Table structure for product
    -- ----------------------------
    CREATE TABLE product (
      id bigint NOT NULL,
      name STRING,
      authors STRING,
      create_time DATETIME
    ) PRIMARY KEY (id) DISTRIBUTED BY HASH (id) BUCKETS 3;
  5. 在kafka上运行命令将MySQL中的product表的变更通过CDC捕获:
    curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" 192.168.100.7:8083/connectors/  -d '{
      "name": "productCDCTask",
      "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        "database.hostname": "192.168.100.10",
        "database.port": "3306",
        "database.user": "your_user",
        "database.password": "your_password",
        "database.server.id": "184001",
        "database.server.name": "demo",
        "database.whitelist": "skyline",
        "database.useTimezone": "true",
    	"database.serverTimeZone": "Asia/Shanghai",
        "time.precision.mode": "connect",
        "decimal.handling.mode": "string",
        "table.whitelist": "skyline.product",
        "database.history.kafka.bootstrap.servers": "192.168.100.7:9092,192.168.100.8:9092,192.168.100.9:9092",
        "database.history.kafka.topic": "dbhistory.demo" 
      }
    }';

创建应用

为了更快地创建自己的flink-jobs应用程序,我们直接下载flink-jobs-quickstart来开始构建我们的第一个flink-jobs应用程序:

  1. 下载flink-jobs-quickstart源代码,并根据自己的情况修改flink-jobs.properties配置文件(这里只需要修改Kafka和StarRocks的配置即可)。
  2. 在flink-jobs-quickstart目录下,使用maven打包:
    mvn clean package -Dmaven.test.skip=true

运行应用

 打包成功后,生成的程序包在target目录下:

  1. flink-jobs-quickstart/target/flink-jobs-quickstart-1.1.3.jarflink-jobs-quickstart/target/lib/*上传到flink所在服务器,例如上传到/opt/flink-jobs-quickstart/flink-jobs-quickstart-1.1.3.jar/opt/flink-jobs-quickstart/lib/*。这里需要注意1.1.3为版本号,请根据自己下载的实际版本调整。
  2. flink-jobs-quickstart/target/lib-provided/*包上传到${yourFlinkPath}/lib目录下(${yourFlinkPath}为你的flink所在目录)。
  3. 启动flink集群,如果此前已经启动,由于${yourFlinkPath}/lib目录发生了变化,重启以确保所有依赖包被重新加载。
  4. 通过命令行直接运行同步任务:
    ${yourFlinkPath}/bin/flink run /opt/flink-jobs-quickstart/flink-jobs-quickstart-1.1.3.jar "{"operates":[{"from":"kafka","primaryKey":"id","table":"product","to":"starrocks","topic":"demo.skyline.product","type":"DataSync"}]}"
  5. 可以尝试对MySQL中的product表进行增删改操作,新的数据将很快同步到StarRocks中。

系统集成

直接通过命令行提交flink-jobs任务并不是终点,因为它使用起来比较繁琐,且不利于管理和维护。最理想的状态应该是直接通过系统界面来进行任务的管理,包括任务配置、启动、停止、重启、监控等。

通过flink-jobs-launcher+Flink REST API可以实现这种目的。首先,利用flink-jobs-launcher提交flink-jobs应用程序会更加便捷,可以采用可读性更好的XML来配置作业,还可以将flink作业集成到现有Java系统中。其次,通过定时调度调用Flink REST API可以将Flink任务监控集成到系统中。详情请移步flink-jobs-launcherflink-jobs-launcher-quickstart 

玩转flink

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
THE END
分享
二维码
< <上一篇

)">
下一篇>>