01业务挑战
数据服务与数据分析场景是数据应用的两个大方向，数据工作从业者可能会遇到以下问题：
数据服务
- 烟囱式开发模式:每来一个需求开发一个数据服务，数据服务无法复用，难以平台化，技术上无法积累。
- 服务维护难度大:当开发可大量数据服务后，后期维护是大问题。尤其是618、双11大促期间，在没有统一的运维监控、限流降级、业务容灾方案的情况下，一个人维护上百个数据服务是很痛苦的，也造成了很大的安全隐患。
- 业务需求量大:数据开发的同学常常会被大量重复枯燥的数据服务开发束缚，大量时间投入在业务数据服务开发中。
数据分析
- 找数据难:用户难以找到自己所想，即便找到名称相近的指标或数据，由于指标定义不明确、不统一，也无法直接使用。
- 用数据难:由于目前数据分布在各个系统中，用户无法用一个系统满足所有的数据需求。特别是一线运营人员，要通过从各个系统导出大量Excel 表格:的方式做数据分析，费时费力，同时也造成数据安全隐患。
- 查询慢:用传统的OLAP引擎，用户跑SQL往往需要几分钟才出结果，大大降低了分析人员的效率。
除了以上问题，数据服务和数据分析系统也无法统一，分析产生的数据结果往往是离线的，需要额外开发数据服务，无法快速转化为线上服务赋能外部系统，使得分析和服务之间难以快速形成闭环。
而且在以往数据加工过程中存储往往只考虑了当时的需求，当后续需求场景扩展，最初的存储引擎可能不适用，导致一份数据针对不同的场景要存储到不同的存储引擎，带来数据一致性隐患和成本浪费问题。
02建设方式
基于以上业务痛点，京东物流运营数据产品团队研发了服务分析一体化平台——Udata。Udata 系统是以 StarRocks 引擎为技术基础实现的。Udata 把数据指标生成的过程抽象出来，用配置的方式低代码化生成数据服务，大大降低了开发复杂性和难度，让非研发同学也可以根据自己的需求配置和发布数据服务。指标的开发时间由之前的一两天缩短为 30 分钟，大大解放了研发力。
平台化的指标管理体系和数据地图的功能，让用户更加直观和方便地查找与维护指标，同时也让指标复用变成可能。在数据分析方面，我们用基于 StarRocks 的联邦查询方案打造了 Udata 统一查询引擎，解决了查询引擎不统一和数据孤岛问题。
同时 StarRocks 提供了强悍的数据查询性能，无论是大宽表还是多表关联查询性能都十分出色。StarRocks 提供数据实时摄入的能力和多种实时数据模型，可以很好支持数据实时更新场景。Udata 系统把分析和服务结合在一起，让分析和服务不再是分割的两个过程，用户分析出有价值的数据后可以立即生成对应的数据服务，让服务分析快速闭环。
改造前的数据流程架构：
实时数据由JDQ(京东日志消息队列，类似ApacheKafka)和 JMQ导入Apache Flink 做实时数据加工，加工后数据写入ClickHouse和ElasticSearch，为数据服务和数据分析提供OLAP查询服务。
离线数据由Apache Spark做个数仓层级加工，APP层数据会同步至MySQL 或ClickHouse 做OLAP查询。
此架构中，在数据服务和数据分析是两个分隔的部分，分析工具由于要跨多数据源和不同的查询语言做数据分析比较困难的，数据服务也是烟囱式开发。
实时链路方面，我们在原有的ClickHouse 和ElasticSearch 基础上引入StarRocks，实现了极速的单表和多表查询能力。
后面我们又以 StarRocks 为基础打造统一查询引擎。统一查询引擎根据京东的业务特点增加了数据源和聚合下推等功能，Udata 在统一查询引擎的基础上，统一了数据分析和数据服务功能。
打造一款数据服务分析一体化系统对查询引擎有比较高的要求，需要同时满足：极速的查询性能、支持联邦查询、实时与离线存储统一。基于这三点要求，接下来，我们就StarRocks 的极速查询性能、我们对联邦查询的改造、实时场景的实践展开讨论。
03业务价值
StarRocks 在联邦查询上支持了多种外表如 ElasticSearch、MySQL、Hive、数据湖等，已经有了很好的联邦查询基础。不过在实际的业务场景中，一些聚合类查询需要从外部数据源拉取数据再聚合，加上这些数据源自身的聚合性能也不错，反而增加了查询时间。
我们的思路是，让这部分擅长聚合的引擎自己做聚合，把聚合操作下推到外部引擎。目前符合这个优化条件的引擎有：MySQL、ElasticSearch、ClickHouse。同时为了兼容更多的数据源，我们还增加了 JSF(京东内部 RPC 服务)/ HTTP 数据源，下面重点介绍下这两部分：
MySQL、ElasticSearch 的聚合下推功能
现在 StarRocks 对于聚合外部数据源的方案是，拉取谓词下推后的全量数据。虽然谓词下推后已经过滤了一部分数据，但是把数据拉取到 StarRocks 再聚合是一个很重的操作，导致聚合时间不理想。我们选择下推聚合操作，让外部表引擎自己做聚合，节省数据拉取时间，同时本地化聚合效率更高。
首先看一个执行计划生成的过程：
- SQL Parse:将SQL文本转换成一个AST(抽象语法树)
- SQL Analyze:基于AST进行语法和语义分析
- SQL Logical Plan:将AST 转换成逻辑计划SQLOptimize:基于关系代数、统计信息、Cost模型，对逻辑计划进行重写、转换，选择出Cost“最低”的物理执行计划
- 生成Plan Fragment:将Optimizer 选择的物理执行计划转换为BE可以直接执行的PlanFragment
我们在步骤 4 中对生成的物理执行计划再次优化，当遇到 ElasticSearch 或者 MySQL 的聚合操作时，会把 ScanNode+AGGNode 的执行计划优化为 QueryNode。QueryNode 为一种特殊的ScanNode，与普通 ScanNode 的区别在于， QueryNode 会把聚合查询请求直接发送到对应外部引擎，而不是 Scan 数据后在本地执行聚合。
其中，用到 ElasticSearch QueryNode 时，我们会在 FE 端就生成 ElasticSearch 查询的 DSL 语句，直接下推到 BE 端查询 。同时在 BE 端，我们实现了 ElasticSearch QueryNode 和 MySQL QueryNode 这两种 QueryNode。我们也为这个 feature 设置了 agg_push_down 的开关，默认是关闭的。
参考下图查询语句，下方是来自 ElasticSearch 和 MySQL 的两个数据源聚合后 Join 查询的 SQL。原先的执行计划需要从数据源 Scan 大量数据后本地聚合，优化后去掉了这部分 Scan 和聚合的过程，直接拿到聚合后的数据。
[图片]
上述 SQL 在优化后生成的执行计划变化如下：
增加 JSF(京东内部 RPC 服务)/ HTTP 数据源
数据服务中可能会涉及到整合外部数据服务和复用原先已开发指标的场景。我们的思路是，把JSF(京东内部 RPC 服务)/ HTTP 也抽象成 StarRocks 的外部表，用户可以通过 SQL 像查询数据库一样访问数据服务。
这样不仅可以复用老的指标，还可以结合其他数据源的数据，生成新的复合指标。我们在 FE 和 BE 端同时增加 JSF 和 HTTP 两种 ScanNode。