TiDB / TiSpark 在易果集团实时数仓中的创新实践

作者简介:罗瑞星,曾就职于前程无忧,参加过 Elasticsearch 官方文档中文翻译工作,现就职于易果集团,担任资深大数据工程师,负责易果集团数据分析架构设计等工作。

项目背景

目前企业大多数的数据分析场景的解决方案底层都是围绕 Hadoop 大数据生态展开的,常见的如 HDFS + Hive + Spark + Presto + Kylin,在易果集团,我们初期也是采取这种思路,但是随着业务规模的快速增长和需求的不断变化,一些实时或者准实时的需求变得越来越多,这类业务除了有实时的 OLTP 需求,还伴随着一些有一定复杂度的 OLAP 的需求,单纯地使用 Hadoop 已经无法满足需求。

现有的准实时系统运行在 SQL Server 之上,通过开发人员编写和维护相应的存储过程来实现。由于数据量不大,SQL Server 能够满足需求,但是随着业务的发展,数据量随之增长,SQL Server 越来越不能满足需求,当数据量到达一定的阶段,性能便会出现拐点。这个时候,这套方案已完全无法支撑业务,不得不重新设计新的方案。

选型评估

在评估初期,Greenplum、Kudu、TiDB 都进入了我们的视野,对于新的实时系统,我们有主要考虑点:

  • 首先,系统既要满足 OLAP 还要满足 OLTP 的基本需求;

  • 其次,新系统要尽量降低业务的使用要求;

  • 最后,新系统最好能够与现有的 Hadoop 体系相结合。

Greenplum 是一套基于 PostgreSQL 分析为主的 MPP 引擎,大多用在并发度不高的离线分析场景,但在 OLTP 方面,我们的初步测试发现其对比 TiDB 的性能差很多。

再说说 Kudu。Kudu 是 CDH 2015年发布的一套介于 Hbase 和 HDFS 中间的一套存储系统,目前在国内主要是小米公司应用的较多,在测试中,我们发现其在 OLTP 表现大致与 TiDB 相当,但是一些中等数据量下,其分析性能相比 TiDB 有一定差距。另外我们的查询目前主要以 Presto 为主,Presto 对接 Kudu 和 PostgreSQL 都是需要考虑兼容性的问题,而 TiDB 兼容 MySQL 协议,在应用初期可以直接使用 Presto-MySQL 进行统一查询,下一步再考虑专门开发 Presto-TiDB。

另外,我们希望未来的实时系统和离线系统能够通用,一套代码在两个系统中都能够完全兼容,目前 Tispark 和 SparkSQL 已经很大程度上实现了这点,这支持我们在以后离线上的小时级任务可以直接切换到 TiDB上,在 TiDB 上实现实时业务的同时,如果有 T+1 的需求也能够直接指 HDFS 即可,不用二次开发,这是 Kudu 和 GP 暂时实现不了的。

最后,TiSpark 是建立在 Spark 引擎之上,Spark 在机器学习领域里有诸如 Mllib 等诸多成熟的项目,对比 GP 和 Kudu,算法工程师们使用 TiSpark 去操作 TiDB 的门槛非常低,同时也会大大提升算法工程师们的效率。

经过综合的考虑,我们最终决定使用 TiDB 作为新的实时系统。同时,目前 TiDB 的社区活跃度非常好,这也是我们考虑的一个很重要的方面。

TiDB 简介

在这里介绍一下 TiDB 的相关特性:TiDB 是基于 Google Spanner/F1 论文启发开源的一套 NewSQL 数据库,它具备如下 NewSQL 核心特性:

  • SQL支持 (TiDB 是 MySQL 兼容的)

  • 水平线性弹性扩展

  • 分布式事务

  • 数据强一致性保证

  • 故障自恢复的高可用

同时,TiDB 还有一套丰富的生态工具,例如:快速部署的 TiDB-Ansible、无缝迁移 MySQL 的 Syncer、异构数据迁移工具 Wormhole、以及 TiDB-Binlog、Backup & Recovery 等。

SQL Server 迁移到 TiDB

由于我们公司的架构是 .NET + SQL Server 架构,所以我们无法像大多数公司一样去使用 MySQL Binlog 去做数据同步,当然也就无法使用 TiDB 官方提供的 Syncer 工具了。因此我们采用了 Flume + Kafka 的架构,我们自己开发了基于 Flume 的 SQL Server Source 去实时监控 SQL Server 数据变化,进行捕捉并写入 Kafka 中,同时,我们使用 Spark Streaming 去读取 Kafka 中的数据并写入 TiDB,同时我们将之前 SQL Server 的存储过程改造成定时调度的 MySQL 脚本。

图:SQL Server 数据迁移到 TiDB

TiDB 前期测试

在测试初期,我们采用 TiDB 的版本为 RC4,在测试过程中曾经在同时对一张表进行读写时,出现 Region is stale 的错误,在 GitHub 上提出 Issue 后,TiDB 官方很快在 Pre-GA 版本中进行了修复。在测试环境,我们是手动通过二进制包的形式来部署 TiDB ,虽然比较简单,但是当 TiDB 发布 GA 版本之后,版本升级却是一个比较大的问题,由于早期没有使用 TiDB-ansible 安装,官方制作的升级脚本无法使用,而手动进行滚动升级等操作非常麻烦。

由于当时是测试环境,在听取了 TiDB 官方的建议之后,我们重新利用 TiDB 官方提供的 TiDB-ansible 部署了 TiDB 的 GA 版本。只需要下载官方提供的包,修改相应的配置,就能完成安装和部署。官方也提供了升级脚本,能够在相邻的 TiDB 版本之前完成无缝滚动升级。同时 TiDB-ansible 默认会提供 Prometheus + Grafana 的监控安装,官方提供了非常丰富完善的 Grafana 模板,省去了运维很多监控配置的工作量,借着 TiDB 部署监控的契机,我们也完成了诸如 Redis,RabbitMQ,Elasticsearch 等很多应用程序的监控由 Zabbix 往 Prometheus 的迁移。这里需要注意的是,如果是用官方提供的部署工具部署 Prometheus 和 Grafana,在执行官方的停止脚本时切记跳过相应的组件,以免干扰其他程序的监控。

TiDB 上线过程

在 10 月中旬,随着新机器的采购到位,我们正式将 TiDB 部署到生产环境进行测试,整个架构为 3 台机器,3TiKV+3PD+2TiDB 的架构。在生产环境中的大数据量场景下,遇到了一些新的问题。

首先遇到的问题是 OLTP 方面,Spark Streaming 程序设置的 5 秒一个窗口,当 5 秒之内不能处理完当前批次的数据,就会产生延迟,同时 Streaming 在这个批次结束后会马上启动下一个批次,但是随着时间的积累,延迟的数据就会越来越多,最后甚至延迟了 8 小时之久;另一方面,由于我们使用的是机械硬盘,因此写入的效率十分不稳定,这也是造成写入延迟的一个很主要的因素。

出现问题之后我们立即与 TiDB 官方取得联系,确认 TiDB 整体架构主要基于 SSD 存储性能之上进行设计的。我们将 3 台机器的硬盘都换成了 SSD;与此同时,我们的工程师也开发了相应的同步程序来替代 Spark Streaming,随着硬件的更新以及程序的替换,写入方面逐渐稳定,程序运行的方式也和 Streaming 程序类似,多程序同时指定一个 Kafka 的 Group ID,同时连接不同机器的 TiDB 以达到写入效率最大化,同时也实现了 HA,保证了即使一个进程挂掉也不影响整体数据的写入。

在 OLTP 优化结束之后,随之而来的是分析方面的需求。由于我们对 TiDB 的定位是实时数据仓库,这样就会像 Hadoop 一样存在很多 ETL 的流程,在 Hadoop 的流程中,以 T+1 为主的任务占据了绝大多数,而这些任务普遍在凌晨启动执行,因此只能用于对时间延迟比较大的场景,对实时性要求比较高的场景则不适合,而 TiDB 则能很好的满足实时或者准实时的需求,在我们的业务场景下,很多任务以 5-10 分钟为执行周期,因此,必须确保任务的执行时长在间隔周期内完成。

我们取了两个在 SQL Server 上跑的比较慢的重要脚本做了迁移,相比于 SQL Server/MySQL 迁移至 Hadoop,从 SQL Server 迁移至 TiDB 的改动非常小,SQL Server 的 Merge 操作在 TiDB 里也通过 replace into 能够完成,其余一些 SQL Server 的特性,也能够通过 TiDB 的多行事务得以实现,在这一方面,TiDB 的 GA 版本已经做的非常完善,高度兼容 MySQL,因此迁移的成本非常小,从而使我们能够将大部分精力放在了调优方面。

在脚本迁移完毕之后,一些简单的脚本能够在秒级完成达到了我们的预期。但是一些复杂的脚本的表现在初期并没表现出优势,一些脚本与 SQL Server 持平甚至更慢,其中最大的脚本 SQL 代码量一共 1000 多行,涉及将近 20 张中间表。在之前的 SQL Server 上,随着数据量慢慢增大,每天的执行时长逐渐由 1-2 分钟增长到 5-6 分钟甚至更久,在双11当天凌晨,随着单量的涌入和其他任务的干扰延迟到 20 分钟甚至以上。在迁移至 TiDB 初期,在半天的数据量下 TiDB 的执行时长大致为 15 分钟左右,与 SQL Server 大致相同,但是并不能满足我们的预期。我们参考了 TiDB 的相关文档对查询参数做了一些调优,几个重要参数为:tidb_distsql_scan_concurrency,tidb_index_serial_scan_concurrency,tidb_index_join_batch_size(TiDB 提供了很好的并行计算能力)。经过验证,调整参数后,一些 SQL 能够缩短一倍的执行时间,但这里依旧不能完全满足我们的需求。

引入 TiSpark

随后,我们把目光转向了 TiDB 的一个子项目 TiSpark,用官网的介绍来讲 TiSpark 就是借助 Spark 平台,同时融合 TiKV 分布式集群的优势,和 TiDB 一起解决 HTAP 的需求。TiDB-ansible 中也带有 TiSpark 的配置,由于我们已经拥有了 Spark 集群,所以直接在现有的 Spark 集群中集成了 TiSpark。虽然该项目开发不久,但是经过测试,收益非常明显。

TiSpark 的配置非常简单,只需要把 TiSprak 相关的 jar 包放入 Spark 集群中的 jars 文件夹中就能引入 TiSpark,同时官方也提供了 3 个脚本,其中两个是启动和停止 TiSpark 的 Thrift Server,另一个是提供的 TiSpark 的 cli 客户端,这样我们就能像使用 Hive 一样使用 TiSpark 去做查询。

在初步使用之后,我们发现一些诸如 select count(*) from table 等 SQL 相比于 TiDB 有非常明显的提升,一些简单的 OLAP 的查询基本上都能够在 5 秒之内返回结果。经过初步测试,大致在 OLAP 的结论如下:一些简单的查询 SQL,在数据量百万级左右,TiDB 的执行效率可能会比 TiSpark 更好,在数据量增多之后 TiSpark 的执行效率会超过 TiDB,当然这也看 TiKV 的配置、表结构等。在 TiSpark 的使用过程中,我们发现 TiSpark 的查询结果在百万级时,执行时间都非常稳定,而 TiDB 的查询时间则会随着数据量的增长而增长(经过与 TiDB 官方沟通,这个情况主要是因为没有比较好的索引进行数据筛选)。针对我们的订单表做测试,在数据量为近百万级时,TiDB 的执行时间为 2 秒左右,TiSpark 的执行时间为 7 秒;当数据量增长为近千万级时,TiDB 的执行时间大致为 12 秒(不考虑缓存),TiSpark 依旧为 7 秒,非常稳定。

因此,我们决定将一些复杂的 ETL 脚本用 TiSpark 来实现,对上述的复杂脚本进行分析后,我们发现,大多数脚本中间表很多,在 SQL Server 中是通过 SQL Server 内存表实现,而迁移至 TiDB,每张中间表都要删除和插入落地,这些开销大大增加了执行时长(据官方答复 TiDB 很快也会支持 View、内存表)。在有了 TiSpark 之后,我们便利用 TiSpark 将中间表缓存为 Spark 的内存表,只需要将最后的数据落地回 TiDB,再执行 Merge 操作即可,这样省掉了很多中间数据的落地,大大节省了很多脚本执行的时间。

在查询速度解决之后,我们发现脚本中会有很多针对中间表 update 和 delete 的语句。目前 TiSpark 暂时不支持 update 和 delete 的操作(和 TiSpark 作者沟通,后续会考虑支持这两个操作),我们便尝试了两种方案,一部分执行类似于 Hive,采用 insert into 一张新表的方式来解决;另外一部分,我们引入了 Spark 中的 Snappydata 作为一部分内存表存储,在 Snappydata 中进行 update 和 delete,以达到想要的目的。因为都是 Spark 的项目,因此在融合两个项目的时候还是比较轻松的。

最后,关于实时的调度工具,目前我们是和离线调度一起进行调度,这也带来了一些问题,每次脚本都会初始化一些 Spark 参数等,这也相当耗时。在未来,我们打算采用 Spark Streaming 作为调度工具,每次执行完成之后记录时间戳,Spark Streaming 只需监控时间戳变化即可,能够避免多次初始化的耗时,通过 Spark 监控,我们也能够清楚的看到任务的延迟和一些状态,这一部分将在未来进行测试。

TiDB 官方支持

在迁移过程中,我们得到了 TiDB 官方很好的支持,其中也包括 TiSpark 相关的技术负责人,一些 TiSpark 的 Corner Case 及使用问题,我们都会在群里抛出,TiDB 的官方人员会非常及时的帮助我们解决问题,在官方支持下,我们迁移至 TiSpark 的过程很顺利,没有受到什么太大的技术阻碍。

实时数仓 TiDB / TiSpark

在迁移完成之后,其中一条复杂的 SQL,一共 Join 了 12 张表(最大表数量亿级,部分表百万级),在平时小批量的情况下,执行时间会在 5 分钟左右,我们也拿了双 11 全量的数据进行了测试,执行时间在 9 分钟以上,而采用了 TiSpark 的方式去执行,双 11 全量的数据也仅仅花了 1 分钟,性能提升了 9 倍。整个大脚本在 SQL Server 上运行双 11 的全量数据以前至少要消耗 30 分钟,利用 TiDB 去执行大致需要 20 分钟左右,利用 TiSpark 只需要 8 分钟左右,相对 SQL Server 性能提升 4 倍,也就是说,每年数据量最高峰的处理能力达到了分钟级,很好的满足了我们的需求。

最后,不管是用 TiDB 还是用 TiSpark 都会有一部分中间表以及与原表进行 Merge 的操作,这里由于 TiDB 对事务进行的限制,我们也采用以万条为单批次进行批量的插入和 Merge,既避免了超过事务的报错又符合 TiDB 的设计理念,能够达到最佳实践。

有了 TiSpark 这个项目,TiDB 与 Hadoop 的生态体系得到进一步的融合,在没有 TiSpark 之前,我们的系统设计如下:

图:多套数仓并存

可以发现,实时数仓与 T+1 异步数仓是两个相对独立的系统,并没有任何交集,我们需要进行数据实时的同步,同时也会在夜晚做一次异步同步,不管是 Datax 还是 Sqoop 读取关系型数据库的效率都远远达不到 TiSpark 的速度,而在有了 TiSpark 之后,我们可以对 T+1 异步数仓进行整合,于是我们的架构进化为如下:

图:TiDB / TiSpark 实时数仓平台

这样就能够利用 TiSpark 将 TiDB 和 Hadoop 很好的串联起来,互为补充,TiDB 的功能也由单纯的实时数仓变成能够提供如下几个功能混合数据库:

  1. 实时数仓,上游 OLTP 的数据通过 TiDB 实时写入,下游 OLAP 的业务通过 TiDB / TiSpark 实时分析。

  2. T+1 的抽取能够从 TiDB 中利用 TiSpark 进行抽取。

+ TiSpark 速度远远超过 Datax 和 Sqoop 读取关系型数据库的速度;

+ 抽取工具也不用维护多个系统库,只需要维护一个 TiDB 即可,大大方便了业务的统一使用,还节省了多次维护成本。

+ TiDB 天然分布式的设计也保证了系统的稳定、高可用。
  1. TiDB 分布式特性可以很好的平衡热点数据,可以用它作为业务库热点数据的一个备份库,或者直接迁入 TiDB 。

上面这三点也是我们今后去努力的方向,由此可见,TiSpark 不仅对于 ETL 脚本起到了很重要的作用,在我们今后的架构中也起到了举足轻重的作用,为我们创建一个实时的统一的混合数据库提供了可能。

与此同时,我们也得到 TiDB 官方人员的确认,TiDB 将于近期支持视图、分区表,并会持续增强 SQL 优化器,同时也会提供一款名为 TiDB Wormhole 的异构平台数据实时迁移工具来便捷的支持用户的多元化迁移需求。我们也计划将更多的产品线逐步迁入 TiDB。

总结

同时解决 OLAP 和 OLTP 是一件相当困难的事情,TiDB 和 TiSpark 虽然推出不久,但是已经满足很多应用场景,同时在易用性和技术支持上也非常值得称赞,相信 TiDB 一定能够在越来越多的企业中得到广泛应用。

分享到微信

打开微信,使用 “扫一扫” 即可将网页分享至朋友圈。