从 Spark 到 StarRocks：实现58同城湖仓一体架构的高效转型

作者：王世发，吴艳兴等，58同城数据架构部

导读：

本文介绍了58同城在其数据探查平台中引入StarRocks的实践，旨在提升实时查询性能。在面对传统Spark和Hive架构的性能瓶颈时，58同城选择StarRocks作为加速引擎，增强数据湖的分析能力。在迁移过程中，团队克服了多个兼容性问题，并对系统进行了稳定性和易用性的改进，特别是在Java UDF支持和SQL黑名单功能方面。

项目实施一年多以来，58同城取得了显著的成果：

日均迁移SQL数量约 6.5 万条 ，查询成功率稳定在 98% 以上 。

查询性能较迁移前提升 20 倍以上 ，平均查询时间缩短至 3.3 秒 ，P90查询时间为 5 秒 。

1 背景

“数据探查平台”是 58集团统一的 SQL 开发平台，旨在提供一个便捷的环境，让用户能够轻松编写、调试和执行 SQL 查询，并实时查看结果，每天超过10万+SQL 运行，包括 ETL，AdHoc 等场景，针对存储在 HDFS 上的海量数据查询，其底层执行引擎是由我们数据架构部门来提供的技术支持。

1.1 执行引擎原有架构

在引入 StarRocks 引擎之前，针对存储在 HDFS 上的海量数据查询，底层执行引擎的整体架构如下：

在该架构下，大部分的探查 SQL 是通过 Spark 来执行的，少部分查询会路由到 Hive，查询响应时间整体在分钟级，随着公司对降本增效需求的日益增长，这已经无法满足即席查询（Ad-Hoc）对快速响应时间的要求，因此 Ad-Hoc 查询加速就成为亟待解决的问题。

1.2 探查场景引入 StarRocks 引擎

经过深入调研与评估后，我们最终选择了 StarRocks 作为我们数据探查场景的加速引擎，主要基于以下几个原因：

统一数据湖分析能力： 我们只需要简单的创建一个 Hive Catalog，即可实现无缝衔接查询 Hive 表的数据，无需经过复杂的数据预处理。
MPP 框架与向量化执行引擎： 可以大大提升 Ad-Hoc 查询速度，综合官方基准测试结果及我们内部 POC 测试结果，使用 StarRocks 引擎，可以将查询性能提升 10 倍以上。
架构简洁： StarRocks 架构简洁，运维成本较低。

引入 StarRocks 引擎后，数据探查执行引擎的整体架构如下：

当有查询请求到来时，Kyuubi 会首先根据路由规则判断是否可以使用 StarRocks 执行。如果查询满足 StarRocks 的执行要求，我们将优先使用 StarRocks 进行处理。如果在 StarRocks 执行期间遇到异常情况，系统会自动降级到 Spark 执行，以确保用户的查询结果不会受到影响。

我们的整体目标是尽可能多地将数据探查 SQL 透明地迁移到 StarRocks 引擎，并且该过程对用户来说是无感知的，下文将围绕探查 SQL 迁移到 StarRocks 引擎过程中，遇到的一系列问题及其解决方案展开详细讲解。

2 StarRocks 兼容 Spark 改造

在推进探查 SQL 迁移过程中，遇到的最主要的问题就是 StarRocks 与 Spark 查询结果不一致的问题。为了确保迁移过程对用户透明无感知，我们就需要对 StarRocks 进行改造，使其能够兼容 Spark的执行逻辑。

虽然不同查询引擎在架构上存在差异，但概括起来 SQL 语句的执行流程无外乎如下几个步骤：

而在我们迁移过程中遇到的 StarRocks 与 Spark 不兼容问题，基本上贯穿了上面整个执行流程，下面我将按照上述执行流程，依次讲解每个环节中我们发现并解决的不兼容问题。

2.1 语法解析阶段

语法解析阶段主要工作是将查询语句转化成语法树，在这个阶段我们遇到的主要问题就是语法不兼容问题，概括起来包括以下两类问题：

语法不兼容问题

在迁移过程中，语法不兼容问题是比较常见的问题。对于一些相对容易解决的问题，我们通过直接修改 StarRocks 源码来向 Spark 兼容。例如，Spark 对表别名的大小写不敏感，而 StarRocks 则是大小写敏感的。针对这种情况，我们对 StarRocks 源码进行了相应的修改。

对于另一部分语法不兼容问题，我们是通过在 FE 端集成 SQLGlot 插件修复的，该插件能够实现查询语句在不同的 SQL 方言之间相互转换。例如，一些标识符如 key、show、system、group 等，在 Spark 中是可以在 SQL 语句中直接使用的，但在 StarRocks 中会被识别为关键字，无法直接使用。通过 SQLGlot 插件，可以将 SQL 语句中的这些标识符都加上反引号（`），从而使其在 StarRocks 中可以正常执行。

语法不支持问题

对于 Spark 的一些高阶语法，如 LATERAL VIEW 语法、GROUP BY ... WITH CUBE 语法以及 GROUP BY ... GROUPING SETS(...) 语法，StarRocks 社区版本目前尚不支持，但是在我们的线上业务中，这部分语法的使用还是很频繁的。针对这些问题，我们扩展了 StarRocks 的能力，使其能够支持这些语法。

2.2 元数据绑定阶段

在元数据绑定阶段，主要工作是将 SQL 语句中的标识符（如表名、列名等）与数据库的实际元数据进行关联。在这个阶段，我们遇到的主要问题是由于 StarRocks 的 Hive Catalog 默认开启了元数据的缓存，导致的查询结果不一致问题。

例如，当 Hive 表的某个分区数据被重跑后，StarRocks 在一段时间内无法感知到分区元数据的变更，从而导致查询结果不一致。

针对这一问题，经过综合分析后，我们最终决定关闭了 Hive Catalog 的所有元数据缓存功能，这一决定看似简单粗暴，但主要基于以下两点考虑：

我们当前所做的工作是将一部分原来由 Spark 执行的查询迁移到 StarRocks 上执行，而 Spark 本身也并不缓存 Hive 表的元数据，因此即使关闭了 Hive Catalog 的元数据缓存功能，也并不会增加 Hive MetaStore 的整体访问量。
当前我们线上业务的查询并发并不高，缓存元数据并不会带来显著的收益。相反，如果开启了元数据缓存，并且查询了一些分区数很多的 Hive 表（一般为多级分区表），后续元数据缓存功能会周期性的刷新缓存中的元数据，这样反而会增加 Hive MetaStore 的负担。

2.3 查询优化阶段

在查询优化阶段，主要工作是基于 RBO（基于规则的优化）和 CBO（基于成本的优化）对执行计划进行重写，以获取最优的执行计划，提高查询性能。在这个阶段，遇到的主要问题是隐式转换问题。

在 RBO 规则中，有一类隐式转换规则，可以在优化查询时自动进行数据类型转换。例如，假设有一个分区表 t1，分区字段 dt 为字符串类型。如果用户在查询 t1 表时使用数值类型的分区过滤条件，如 where t1.dt = 20241201，那么针对这种不规范的用法，Spark 和 StarRocks 都会使用各自的隐式转换规则进行数据类型转换。

针对这类问题，我们系统的梳理了 Spark 和 StarRocks 在各类表达式中的隐式转换规则，并将StarRocks 的隐式转换规则与 Spark 进行了兼容，基本上彻底解决了这一类问题。

2.4 查询执行阶段

在查询优化阶段结束后，会生成一个由各种算子构成的查询计划树，在查询执行阶段，主要工作是在执行端执行这一系列的算子，例如 Scan 算子负责从存储层读取数据，Expr 算子负责进行表达式计算。在这个阶段遇到的兼容性问题也是最多的，概括起来主要包括如下两类问题：

text 格式的 Hive 表兼容性问题

在我们的生产环境中，有相当一部分 Hive 表使用的是 text+lzo 存储格式。针对这种存储格式，StarRocks 支持的并不完善，例如，早期 StarRocks 版本不支持查询 lzo 压缩的 text 格式的 Hive 表。为了解决这个问题，我们与 StarRocks 社区积极合作，扩展了 StarRocks 对 text+lzo 存储格式的支持。

另外，对于 text 格式的 Hive 表，StarRocks 社区版本也不支持查询 Map 类型的字段，我们也扩展了StarRocks 的能力，使其能够支持对 Map 类型字段的查询。

除此之外，我们还遇到了一些其他的兼容性问题，例如：

hive 表字段分隔符问题： 在某些特殊情况下，StarRocks 在处理 Hive 表字段分隔符时与 Spark 存在不兼容的情况。
临时文件处理问题： StarRocks 在查询 Hive 表时没有忽略存储目录下的临时文件。
空文件处理问题： StarRocks 在解压缩空文件时会抛出异常。

针对这些问题，我们都一一进行了修复，确保了 StarRocks 能够正确处理这些特定的场景。

函数不兼容问题

函数不兼容问题也是我们在迁移过程中花费最多精力处理的问题。具体又可以细分为以下两类：

Spark 与 StarRocks 都有相同功能的函数，但是函数名称不同

这类函数处理起来比较简单，只需要在生成执行计划时，将 Spark 中的函数映射到 StarRocks 具有相同功能的函数即可。

Spark 支持但 StarRocks 不支持或不兼容的函数

对于这类问题，我们采取了两种方式进行处理：

逻辑简单的函数 ：我们通过借助 StarRocks Java UDF 功能，创建 UDF 并在生成执行计划时将这些函数映射到自己实现的 Java UDF 函数来解决。
逻辑复杂的函数 ：我们直接修改 StarRocks 相关函数的代码，使其兼容 Spark 函数的处理逻辑。

通过以上方法，我们共计解决了 40 多个不兼容的函数，主要涉及日期处理，字符串处理，正则匹配，聚合函数等函数，基本上彻底解决了生产环境中存在的函数不兼容问题。

3 实践经验总结

在使用 StarRocks 的过程中，我们从实践中总结出了关于性能、稳定性和易用性的关键经验。

3.1 性能

我们的 StarRocks 集群开启了 Data Cache 功能，以提升查询性能。整体表现令人满意，但在个别场景中，查询性能受到 HDFS DataNode 慢节点问题的影响，出现了性能长尾现象。

针对这一问题，我们采用了自研的 HDFS 功能。当系统检测到某个 DataNode 响应缓慢时，会自动切换至其他副本读取数据。为实现这一功能，我们替换了 StarRocks 依赖的 HDFS JAR 包。经过优化后，系统的 P99 查询性能提升了 25%，效果显著。

3.2 稳定性

在项目初期，我们使用的是 StarRocks 3.0 版本，我们遇到了一些挑战，例如 FE 卡死或 BE 节点偶发性异常等。通过与 StarRocks 社区的积极交流，我们获得了社区的大力支持，并在升级至 StarRocks 3.2 后，整体系统的稳定性有了明显的提升。

其中，一个值得分享的案例是关于 CBO 统计信息的问题。在查询一张包含 3565 列的大宽表时，CBO 优化器会生成一个较大的 SQL 来获取统计信息，这可能导致 FE 内存占用过高，影响集群的正常运行。

针对这一情况，我们优化了相关逻辑，跳过了部分非必要的统计信息查询。这一改进有效降低了查询负载，同时进一步提升了系统的稳定性。

注意：稳定性问题在最新版本已优化。

3.3 易用性

为提升系统的操作便利性，我们对以下功能进行了优化：

Java UDF 支持从 HDFS 下载 JAR 包

增强了 UDF 功能，使 JAR 包可通过 HDFS 下载，简化了运维流程。

SQL 黑名单持久化

改进黑名单存储方式，实现多节点同步和持久化，降低了运维成本。

4 StarRocks 上云

4 .1 背景

为了进一步推进公司的降本增效战略，我们部门与 58 云平台团队进行通力合作，决定将一部分大数据组件迁移到 58 云平台。而对于 StarRocks 本身来说，数据探查场景主要借助其数据湖分析能力查询 Hive 表数据，集群本身是无状态的，并不会存储数据，天然就适合上云，因此我们决定将数据探查场景的 StarRocks 集群上云。

然而，由于各种情况的限制，上云使用的宿主机每台只能提供最多 5CORE 15GB 的资源。这给 StarRocks 的上云之路带来了一些挑战。

4 .2 上云架构

上图为 StarRocks 云集群的整体架构：

由于每个容器使用的资源限制在 5CORE 15GB，而 FE 节点需要存储集群的元数据，对内存资源要求较高，因此FE节点我们并没有上云，仍然使用物理机部署。
BE 集群只是为了存储审计日志表数据，需要开启云集群本地存储，只部署了少量实例。
CN 集群是主要的计算节点，无状态，支持快速的扩缩容。

4 .3 面临的挑战

上云过程中遇到的最大挑战就是容器内存资源不足，这导致 CN 节点容器频繁发生 OOM，被操作系统杀死。为了解决这一问题，我们采取了以下措施：

1 设置资源组隔离及查询队列

通过设置资源组隔离和查询队列，控制查询并发，确保资源的合理利用。

2 开启中间结果落盘功能

开启中间结果落盘功能，将部分计算结果暂存到磁盘上，从而降低内存消耗。

3 减少 CN 进程执行 线程数

CN 进程中有一些与执行线程数相关的参数，默认值通常设置为机器 CPU 的核数。而在容器环境下，CN 进程识别出的 CPU 核数是宿主机整体的 CPU 核数。因此，我们需要根据容器的实际资源配置手动调整这些参数。

4 限制 CN 进程 JVM 内存

在使用 StarRocks 数据湖分析能力查询 Hive 表场景下，有一部分功能是通过 JNI 来实现的，例如读取 HDFS 文件，还有就是一些 JAVA UDF 的使用。然而，由于 JVM 默认堆内存的上限是根据物理机的内存自动设置的，这种默认行为并不适配容器化环境。因此，需要通过设置 CN 进程的 JAVA_OPTS 参数 来限制 JVM 内存的使用，从而确保 CN 进程 的整体内存消耗不会超出容器的内存限制。

通过采取以上一系列措施，基本上解决了因容器内存溢出导致的系统不稳定问题。