亚马逊AWS官方博客

StarRocks 存算分离集群 TB 级迁移实践分享

背景

某光伏新能源企业希望将其海外业务站点搬到 AWS。客户使用 StarRocks on HAQM S3 实现存算分离架构，以支持其大数据量级的多维分析、实时分析、高并发分析等多种数据分析场景。

故，在此次项目中，客户非常重视 StarRocks 集群的迁移方案，原因有五：

  1. 客户的 StarRocks 集群保存了客户的核心业务数据，数据量较大，总计超过 60TB+，并且客户需要进行存量和增量迁移，对切割窗口有严格的要求。
  2. 客户要求迁移要尽量减少对线上源集群的影响，客户的 API 接口熔断阈值相对较低。
  3. 客户的 StarRocks 集群采用了存算分离部署模式，存算分离集群 StarRocks 官方并没有提供数据迁移工具。
  4. 客户的 StarRocks 集群使用了一个较老的版本（3.1.8），客户根据自己业务的需要，对源码也进行一些修改。
  5. 客户目前没有看到 StarRocks 在这个级别数据量级和版本的成功迁移案例。

基于上述要求，我们尝试了一系列迁移方案，并在客户的实验环境成功通过客户验证，进一步赢得客户信任。

需求分析

客户由于降本的诉求，使用了 StarRocks 的存算分离版本，底层数据的存储在对象存储中。相比使用块存储，使用对象存储带来了非常大的成本优势。但是，这为我们迁移带来了第一个困难就是官方的的迁移工具 StarRocks Cross-Cluster Data Migration Tools 仅支持 StarRocks 存算一体集群迁出，不支持以 StarRocks 存算分离集群做为源集群进行数据迁移。

尝试方案一

基于对其他存算分离的大数据产品的理解，例如 Redshift、Hive、Hbase、Hudi 等产品，大致都是海量数据存放在 HAQM S3 上，以及有一个对应的元数据管理。针对基于 HAQM S3 协议的对象存储数据，使用 Data Transfer Hub (DTH) 是一个不二的选择。

所以，我们最初的想法是通过 HAQM DTH 把源 StarRocks 存算分离集群的底层数据，同步到目标集群的 HAQM S3 存储桶中，并通过重建元数据的方式，构建目标集群。

这个过程遇到以下几个难点：

  1. HAQM S3 的分区特性和源 StarRocks 的底层对象存储不一致。由于客户 StarRocks 每天较大的数据增量，同时读取频率也较高，所以对底层对象存储的 API Call 较大。HAQM S3 默认每秒的 API Call 数量不能满足客户需求，需要增加分区，以提升 API Call 的上限。所以 StarRocks 官方是通过在桶的第一层级，添加两位的分区前缀增加 HAQM 分区，前缀结构如下：
  2. StarRocks 集群的目录结构，每一级是由集群 id、数据库 id、表 id、分区 id，再到底层数据组成，迁移期间但凡有干扰数据写入，致使某个 id 被占用，将导致平迁的数据无法识别，导致失败。在海量数据下，排查冲突也将极其困难。
  3. StarRocks 集群会不定期在适当的时机进行 Compaction 操作，该操作会合并小文件，这个过程会重新映射底层文件和 Metadata 之间的关系。迁移过程中的目标集群底层的文件变动了，根据这个变动去反向更新 Metadata 比较困难，尤其是在 HAQM S3 上，由于分区前缀的问题，使得问题更加复杂。

再经历了尝试后，觉得上述方案风险较大，迁移后，无法保证数据的完整性和准确性开始考虑基于原生的导出导入功能进行数据迁移。

尝试方案二

导出导入的方案，由于是原厂支持，理论上不会有太大风险，但是我们需要解决如下问题：

  1. 如何解决增量数据的迁移？
  2. 在数据量较大情况下，如何保证数据迁移的稳定可靠？
  3. 数据量较大的情况下，导出任务占用线上源生成集群资源，是否会影响线上业务？

如下是不同版本 StarRocks 不同版本支持的导出导入方式：

由于目前客户使用的版本是 StarRocks 版本是 3.1.x 版本，由于业务的稳定性，短时间不会进行版本升级，所以可供我们选择的导出方案只有使用 EXPORT 命令，将数据导出成 CSV 格式。

使用 EXPORT 命令，带来的问题较多：

  1. 数据由于没有压缩，数据占用空间很大。经过我们测试，我们导出 2 亿行，单行约 300B 的记录，生成的的多个 CSV 文件，总共约 48GB。
  2. EXPORT 导出的最小粒度是一个分区，如果一个分区很大，网络如果不稳定，会出现失败，并且导出任务不会回滚，会残留很多片段数据，造成数据重复。

我们再来看看可供数据导入的方式：

针对 StarRocks 3.1.X 版本并且要适配 EXPORT 导出的数据格式，我们选择了 Borker Load 的方式，原因有二：

  1. 支持从云端对象存储 HAQM S3 导入（TB 级别数据，存放在本地磁盘成本较高）
  2. StarRocks 原生支持，不需要额外依赖

方案设计

如何实现增量数据的迁移

方式一：使用增量表实现增量数据迁移

在迁移开始，使用增量表的方式，解决 EXPORT 命令无法导出增量数据问题。具体方式如下：

  1. 创建一个和被导出表相同的 schema 的表做为该表的增量表。
  2. 业务端对原表和增量表在导出开始的时候，进行双写。
  3. 原表导出结束后，切换导出增量表。
  4. 如果增量表的数据量也较大，需要重复 1~3 的步骤，在导出第一个增量表的同时，将数据双写入第二张增量表。
  5. 最终保证最后的增表能在业务的维护窗口一次性迁移完成。

方式二：使用条件过滤导出增量数据迁移

StarRocks 3.3.X 以后的版本，支持 INSERT INTO FILES 的方式导出数据，该方式支持按条件过滤导出数据。格式如下：

INSERT INTO
FILES(
    "path" = "s3://demo_bucket/path/table001/data",
    "format" = "csv",
    "csv.column_separator"="|",
    "compression" = "uncompressed",
    "target_max_file_size" = "104857600",
    "aws.s3.region" = "us-east-2"
)
SELECT * FROM data_point_val
where receive_time >= "2023-11-04 00:00:00" and receive_time < "2023-11-05 00:00:00";

本方案利用这个特性实现增量数据迁移，具体原理如下：

  1. 使用该方案涉及的迁移工具迁移完存量数据，并记录每个分区导出起始时间。
  2. 该方案的迁移工具会重新对每个分区导出任务添加过滤条件，只导出从上次导出时间开始新插入的数据，再次迁移这些新增数据。
  3. 在最后业务暂停窗口，完成最后一轮的数据补齐。

两种方式的取舍

方式一的优势是减少了多轮对源集群分区的遍历，对源集群的影响较小，但是需要业务端配合实现增量表，方式二会对源集群多轮查询，对源集群影响相对较大。但是无需业务端配合，心智负担较小。

如何减少对源集群的影响

在实践过程中，我们发现对源的影响主要有三块：

  1. 由于是底层是对象存储，所以导出大量数据的时候，对对象存储的 API Request 非常巨大，迁移之前需要在测试环境评估每秒 API Request 的个数，提前开 case 提高访问请求容量，同时控制导出并发和导出时间间隔。
  2. 对集群 CN 节点的 CPU、内存、磁盘 IOPS 影响都有巨大影响，需要控制导出并发和导出时间间隔。
  3. 我们的方案中，定时调用 StarRocks 的 API 获取了上述指标信息，动态控制了导出的并发和时间间隔，减少迁移过程中的心智负担。

如何监控任务的迁移任务状态和数据的一致性

我们在实践过程中，将导出任务按 Partition 拆分，将导入任务按文件拆分，全程使用 HAQM DynamoDB 对任务状态进行监控，对失败的任务进行重试，保证数据至少一次迁移成功。

  1. 如果被迁移的 StarRocks 表具有主键模型，重复的数据导入，并不会造成数据的不一致。
  2. 如果被迁移的表不具备主键模型，我们需要删除当前具有错误任务的分区，重新迁移该分区。
  3. 迁移后，我们也会对数据根据行数，和一些统计指标，对比数据，校验数据的一致性。

如何保证迁移过程的网络稳定性

对象存储之间，我们使用了 DTH，利用对象存储的网络带宽，实现文件传递。DTH 内部对文件进行了完整性检查，加密等，保证了传输过程中的安全稳定，更多详情请查看 Data Transfer Hub (DTH)。

方案架构

方案的架构设计最终如下：

核心组件

  1. Task Manager 负责任务的调度和管理。一般可以部署在 HAQM EC2 或者 HAQM EKS 中。负责处理的任务主要有两种，一种是导出任务，从源 StarRocks 导出数据。一种数导入任务，从对象存储中，把数据导入目标集群。Task Manager 还负责任务的编排，队列管理，状态记录等。
  2. File Event Trigger。一般由 HAQM Lambda 实现， 负责在目标端的对象存储接收到新文件后，负责将文件的状态信息写入 Task Recorder 中，并推送到 Task Queue 中。
  3. Task Queue。一般由 HAQM SQS 实现。记录任务的元信息，包括任务文件位置，数据所属于的分区等。Task Manager 会从 Task Queue 中消费任务，并转化成导入命令。
  4. HAQM DTH。 如果源集群和目标集群之间没有专线，之间通过公网跨云传输任务，在数据量较大的情况下，由于网络抖动，容易出现传输失败，或者部分传输。这种情况下建议使用 DTH。源集群先把数据导入源端对象存储，再使用 DTH 利用对象存储的公网带宽，同步到目标端对象存储。
  5. Object Storage。一般由对象存储实现，AWS 中使用 HAQM S3。

架构说明

  1. Task Manager 读取源集群后，按分区将数据导出到源端对象存储，并更新每个正在导出的任务状态。
  2. HAQM DTH 监听到源对象存储有新文件产生后，将文件传输到目标端对象存储（HAQM S3）。
  3. File Event Trigger 监听到目标端对象存储有新的文件产生后，推送到 Task Queue 中，并记录任务状态。
  4. Task Manager 从 Task Queue中消费任务，并产生导入任务指令，发送给目标端 StarRocks。
  5. Task Manager 监听 StarRocks 导入任务状态，并记录到 Task Recorder。

如果源端到目标端有比较稳定的专线网络，可以指标将数据导出到目标端对象存储，降低架构复杂度。

补充说明

目前方案具体的实现代码参考这里。代码中针对 StarRocks 存算分离版本，根据官方的推荐，使用了不同的导出导入策略。

总结

本篇文章定位构建一种针对 StarRocks 存算分离集群，支持 TB 级别数据规模的迁移方案。我们通过使用 HAQM SQS、HAQM DynamoDB、HAQM DTH 等多种服务，结合 StarRocks 的导出导入功能，实现了一种大规模数据的迁移方案。该方案将数据按分区粒度进行迁移，并通过监控源集群资源监控，动态挑战任务并发度和时间间隔，以减少对源集群的影响。同时该方案通过双写技术，实现了 StarRocks 存算分离集群增量迁移。

本篇作者

许庭新

西云数据解决方案架构师，10+年产品研发和解决方案咨询经验，在电商，互联网金融，智能汽车领域有丰富的实战经验，擅长利用云计算、大数据，AI 等技术挖掘用户底层需求，实现精准运营。

苏志勇

亚马逊云科技迁移解决方案架构师，主要负责企业上云跨云迁移相关的技术支持工作。曾担任研发工程师、解决方案架构师等职位，在 IT 专业服务和企业应用架构方面拥有多年的实践经验。

蔡如海

西云数据解决方案架构师，10+年开发和架构经验，曾就职于知名外企，在媒体、金融等业务领域有丰富的工作经验，擅长云计算、机器学习等技术，并且有丰富的项目管理经验。