Fork me on GitHub

去哪儿网数据同步平台技术演进与实践

图片

作者介绍

井显生,2019年加入去哪儿,现负责国内机票出票、退款、改签核心业务。在领域驱动设计(DDD)、高并发有大量实践经验。

一、前言

去哪儿网国内机票售后是为用户提供退票、改签、航班变动、行程服务、疫情政策等服务的业务。业务场景中有复杂的基于订单、客票、 PNR 、行程、航司等各种维度的查询诉求,这些数据分散存储在不同的 mysql 数据库表中,难以为复杂业务查询场景提供服务。为解决复杂查询场景,我们设计了一套将数据从一个数据源聚合导入到另一个数据源,提供同构或者异构、低延时的、最终一致性的数据同步系统。

去哪儿数据同步平台是把数据从 mysql 同步到 es 中,并且提供统一的查询网关的服务。随着业务的发展, es 主备集群高可用性存在问题,单一节点故障后难以恢复和补数;同步链路不合理,索引间存在相互影响;公司要求升级 es 集群版本等问题凸显。本文将与大家详细探讨数据同步平台 ES 多版本迁移支持、数据同步平台高可用和数据一致性的设计与实践,希望对大家能够有所帮助或启发。

二、数据同步平台介绍

早期去哪儿推出了基于 Databus 的数据同步系统 1.0 。它解决了多表聚合后写入异构数据存储系统(如 es ,hive 等)的问题。为解决 databus 生产者单点问题、canal和数据配置运维困难和查询客户端管理混乱等痛点,数据同步平台做了整体的架构升级。具体的实践,大家可以参考 Qunar 技术沙龙之前的技术博客《数据同步平台重构实践》。去哪儿数据同步平台主要是数据同步模块、数据中台( crab )和管理模块组成,整体架构如下图所示。

图片

2.1 数据同步平台介绍

数据同步模块包括 databus 、 canal 、 inception gate 等多种实现方案,我们从同步配置易用性、事务性和高可用上考虑选择了阿里巴巴开源分布式数据库同步系统( otter )订阅 mysql binlog ,消息队列使用了 kafka ,数据映射过滤规则自主设计的系统 dts 。

  • **otter : ** 基于阿里巴巴开源分布式数据库同步系统改造,主要解决了同步配置页面化、 Databus 生产者单点的问题。同时我们扩展实现了Kafka类型的DataSourceMedia,在otter Load阶段发送消息给Kafka。
  • **dts : ** 设计思路参考 otter 的模式,主要包括 Node 、补数任务和数据库反查组件。Node 是由 SFTL 四个阶段组成, S 是从 kafka 消费数据, F 是过滤数据, T 是重新组装数据(主子表关联、 Array 处理), L 是把数据通过 Crab-client 写入 ES 网关中。

2.2 数据中台

数据中台( crab )对内提供了数据同步写入服务,对外提供了统一的查询服务。实现了以 appcode + es 索引维度的 hystrix 熔断、降级和隔离,通过配置设置了 es 集群的查询流量调度。我们还提供了 crab-client 以 java api 形式屏蔽了dsl复杂细节,降低了业务线接入成本。

  • **crab : ** 是数据中台,提供了 ES 的读写服务,提供统一的鉴权、熔断限流、流量分配等功能。
  • es group: es group 是指一个 index 存在于多个 es 机群中,逻辑上划分为组,为读写提供主备高可用服务。
  • 业务接入方: 业务接入方通过 crab-client 接入对 es 进行查询, crab-client 简化了查询复杂度,方便业务接入。

2.3 管理平台

管理平台对整个数据同步平台来说可以是一个 optional 的环境,只有在第一次启动启动时需要,一定启动了任务后,无论管理平台是否可用,不影响正常功能。管理平台维护了数据同步配置、 dts node 的上下线,管理数据中台 crab 的鉴权、限流和 ES 集群读写流量分配。

  • mysql mmm pxc: 公司内部 mysql 三种架构(单节点mysql、mmm 和 pxc )。由 otter 内置 canal 订阅 binlog ,提供数据同步元数据。
  • manager : 管理工作节点( otter 、 dts 、 crab )运行时的配置,主要包括读写鉴权、 ES 集群分组、 ES 索引维护、 ES 主备集群查询流量分配、dts 数据聚合 node 管理和其他管理功能。
  • watcher: 公司内部监控报警平台,为系统提供全链路的监控和报警。

三、技术演进背景

数据同步平台目前接入了 10+ 业务线、14+个 es index。2021 Q2 数据组要求统一升级 es5.x 到 es7.x ,同时数据同步平台高可用和稳定性问题凸显,主要表现在以下四个方面:

  • ES 主备(任一)机群故障后,整个同步链路不可用,且故障难恢复。
  • 数据库 IP 迁移后,otter 无法自动切换数据库。
  • 全链路监控和降级熔断不清晰,难以发现故障,排查问题难。
  • 同步链路隔离和降级不合理,一个索引同步故障影响了其他索引的同步。

基于上述系统痛点,制定了以下两个目标:

  1. 数据同步平台系统具备灵活的缩扩容能力,把 es 集群平滑缩扩容迁移,升级 es5.x 到 es7.x 。
    这个目标不仅满足数据组需求,还提高了数据同步平台的高可用能力,单一 es 集群故障时可以灵活的配置上下线恢复故障。
  2. 同步全链路高可用优化
    梳理全链路的流量治理和熔断降级机制,建立同步各个阶段的高可用方案,隔离不同索引同步间的相互影响。

四、技术演进实践

4.1 es 集群 7.x 升级和热拔插

crab 是数据中台,提供了 es 集群的读写服务,核心架构如下图所示。manager 配置 es 集群分组和查询流量控制, crab 定时拉取配置信息,对外提供读写操作。7.x 升级和热拔插目标可以拆解为crab网关支持 es5.x 和 es7.x 并行、es 集群热拔插、es 集群灵活补数三个任务项。

图片

4.1.1 es5.x 和 es7.x 并行

es5.x 和 es7.x 在版本上存在差异,主要包括 es rest client 兼容性,REST APIs 兼容性,es 查询返回值的三部分:

  • Elastic rest client

es 客户端目前主要是以下四个,从 es5.x 和 es7.x 多版本支持上看 Elastic java low level rest client兼容所有版本es,综合比较后我们选择使用Elastic java low level rest client。

图片

  • REST APIs

crab网关支持的REST APIs主要包括Search APIs、Document APIs 和 Script APIs

Search APIs 主要支持 Search 、Scroll 对 ES 进行查询操作。主要区别如下表格。es7.x 功能实现需要在 manager 中配置 ES 的版本, crab 在构建请求时识别 es 的版本后构建不同的 endpoint 进行适配。

图片

Document APIs主要支持 Index、 Update 和 Delete , crab 网关 insert 和 update 操作都统一使用 upsert 操作实现。

图片

  • Query DSL 和 Scripting

Elasticsearch 提供了基于 JSON 的完整查询 DSL (Domain Specific Language)来定义查询。es5.x 和 es7.x 版本支持的 dsl 存在差异,针对此场景汇总 crab 中支持的 DSL 语句发现 match 有以下区别;

es nested 类型存储的是父子(1:n)表,更新的时候需要利用 script 解决, es7.x 不再支持 file 的形式指定 script ,需要构建请求时区分 es 版本解决。

图片

  • es5.x 和 es7.x 返回值区别

es5.x 和 es7.x 返回值主要差别在 hits 上, es5.x 可以直接返回查询命中数量, es7.x 返回数量是个结构体。如下表。解决方案是 es7.x 的查询 endpoint 默认添加参数 ?track_total_hits=true,返回值的 relation 就是 eq ,获取到返回值后,提取查询名字数量,修改为 es5.x 结构。

图片

4.1.2 es 集群热拔插方案

es 集群热拔插是解决 es7.x 升级的平滑上线和单一 es 集群故障时机群切换的方案。详细切换步骤明细可见下表格。

图片

es7.x 上线从时序上核心 5 个阶段,如下图:

图片

    1. 执行 reindex 阶段:执行 reindex 将老集群全量数据写入到新集群中,此时新集群中包含的数据是时间段1内的全量数据。
    1. crab 写入阶段:执行步骤 5 ,此时新产生的数据可以写入新集群,新集群中缺失时间段 2 的数据。
    1. 后置 diff 补数:从反查组件中查询全量数据,通过crab网关写入组件写入时间段 2 缺失的数据和时间段 3 遗漏的数据。
    1. 验证数据:验证新老集群时间段数据和抽样 diff 新老集群数据。
    1. 可查询阶段:diff 一致后新集群可以切换查询。

4.1.3 es 集群灵活补数

针对补数有全量、部分补数的需求,我们制定了 reindex 、 canal 移动位点、 diff 补数定时任务三个补数方案方案。

  • reindex:ES 提供了 _reindex 这个 API 供索引进行重建,这种方式可以全量数据重建,具体代码如下:

curl -H "Content-Type: application/json"   -XPOST http://ip:port/_reindex -d'{
    "source": {
        "remote": {
        "host": "http://ip:port"
        },

            "index": "order_info_beta_tts8"
     },
     "dest": {
          "index": "order_info_beta_tts8"
     }
}'

reindex 补数方案适用于上图时间段 1 的补数,可以把集群 1 索引的数据全量导入到指定的新集群。

  • canal 移动位点: 通过修改 canal 的位点信息,重新拉取 binlog 方式进行同步补数,由于 binlog 保存时长的问题,位点只能重新同步最近的数据,修改方式如下图:

图片

canal 位点方案适用于时间段近、集群读写 qps 不高的场景,可以用于时间段 2 的补数方案。

  • diff 补数定时任务: 通过反查组件,反查数据库全量数据方式进行es数据重写。

canal 位点方案适用于时间段近、集群读写 qps 不高的场景,可以用于时间段 2 的补数方案。

  • diff 补数定时任务: 通过反查组件,反查数据库全量数据方式进行 es 数据重写。

图片

diff 补数方案是精确补数方案,由业务方实现反查组件,数据同步平台调用获取数据后通过 es 网关获取索引中的数据,对比不一致的通过 crab 写入组件写入对应的索引集群。diff 方案可以实现精确补数,缺点是业务接入方有业务开发量。目前线上支持国内机票、国际机票等核心业务。

4.2 数据一致性和高可用

要介绍数据同步平台的数据一致性和高可用,首先以国内机票为例,核心同步信息有订单信息(1)、乘机人信息(n)、航段信息(n)、退款记录(n)等,同步到 ES 索引结构如下图所示。订单信息创建了单个文档,乘机人、航段和退款记录等数组以nested data type内部文档存储。

图片

数据同步流程是由 otter 内置 canal 订阅 binlog 后经过 SETL 四个流程后,写入到 kafka 。dts 消费 kafka 数据,经过 SFTL 四个流程后写入 crab ,最终写入 ES 集群中,如下图所示。

图片

  • 数据顺序和最终一致性的保证:

数据最终一致性的保证是由三个层面:1. 数据同步的时候链路上同纬度数据是顺序性的;2. 数据在链路写入失败时,进入失败重试队列;3. 针对重要集群建立 diff 补数任务。

  1. 按照单维度数据的同步顺序性(otter → kafka → dts → crab → es)

otter Load 阶段写 kafka 时 partition ,要保证同业务分配到同一个 partition 。例如国内机票业务发送 kafka 的partition key 是库名 + order_id ,这样可以保证一个订单下的所有binlog被发送到同一个 partition 内。

dts 消费 kafka 的时候是单线程的,处理数据也是有顺序的。通过 dubbo 接口写到 crab 网关时,也是顺序写入 es 中。

  1. 数据在链路写入失败时,进入失败重试队列

dts 到 crab 和 crab 处理数据过程中都可能因为各种原因失败异常,系统捕获到异常后将数据写入 retry kafka 队列, dts 消费 retry 数据后,通过数据库反查组件,查询最新数据写入 crab 网关保证数据最终一致。

  1. 针对重要集群建立 diff 补数任务

针对国内机票和国际机票等重要索引,建立了最新一分钟数据 diff 补数逻辑,定时任务通过反查组件,查询全量一分钟数据,跟 es 数据 diff 比较,不一致的数据写入 crab 网关保证数据最终一致性。

  • 数据同步链路的高可用:

数据同步链路的高可用(同步顺序:otter → kafka → dts → crab → es)如下:

otter:otter以Pipeline(可以理解成一个es索引的同步)运行在不同的 otter node 中,实现了索引间的隔离。同时内置 canal 以主备模式, S 和 L 阶段以主备模式运行在多个 Node 中,实现了单索引的高可用。

kafka:以每个索引一个 topic ,每个 topic 有多个 partition ,每个 partition 有多个副本实现了高可用。

dts:以 kafka 的 topic 为维度生成了多个 Node 消费数据,以线程隔离的方式实现了索引间的隔离。多个 consumer 实现了单索引的高可用,一个 consumer 销毁后其他 consumer 可继续消费数据。

crab:以调用方 appcode+ 索引维度创建了 Hystrix 线程池,不仅隔离了索引间的影响还保证了单索引的承载能力。

es:一个索引保存在多个 es 集群中,由管理系统配置查询分流实现了索引的高可用。

由于数据同步平台出现过 kafka 磁盘空间满、 zk 故障等异常 case ,此时可以通过 4.1.3es 集群灵活补数方式对集群进行部署和恢复服务。

  • 数据同步的优化:
  1. 以国际机票生单为例,一个操作是会产生多个主子表的 binlog ,多个 binlog 会写入到同一个 partition , dts 在消费时可以根据业务特性,以服务单单号为 key ,一批内只保留最后一个 binlog 执行反查和写入,即一批 binlog 中同一个服务单单号不管有几个 binlog ,最后执行反查和写入的只有一次。
  2. otter 数据库主备切换并未开源,根据公司主流 mysql 是 pxc 架构,我们定制实现了数据库主备切换功能。
  3. otter S 阶段默认拉取 10000 条数据,在大表有 DDL 操作时会打满网卡造成故障,我们根据拉取条数变化动态调整拉取设置。

五、总结和未来规划

通过以上方案,顺利完成了预期目标,实现了数据同步平台的全链路监控和全链路高可用。另外 es 集群读写性能有了量级的变化,国内机票 es5.x 迁移 7.x 查询耗时从 68ms 下降到21ms (下图1),写入耗时从 34ms 下降到 6ms (下图2):

图片

图片

系统上线后,在日常维护中出现了几次线上问题,系统的高可用性也经过了考验:

  1. 工单es老集群物理机故障

6月4号系统报警提示数据中台写入工单索引异常量增长,查询异常日志后发现是 es ip 连接失败,咨询 noc 得知是机器过保宕机,通过 es 集群热拔插方案手动下线掉工单老集群读写恢复故障。

  1. kafka 磁盘打满导致同步故障

系统报警发现单索引同步量异常,经排查后发现是 kafka 某台机器磁盘满了导致同步异常,解决完磁盘问题后canal移动位点的形式对数据进行补偿恢复服务。

数据同步平台在同步链路中实现了高可用的方案,但是部分场景是需要人工排查异常后解决。随着更多业务线的接入,对我们的可用性、易用性、可扩展性和低成本接入上面提出更高的要求。未来规划重点包括 dts 数据聚合配置化、故障自动迁移等。

六、招募贴

去哪儿网从实习生到技术专家,多岗位在招,hc多多,对以上感兴趣的小伙伴快来联系我们,驼厂欢迎大家的加入。

图片


本文地址:https://www.6aiq.com/article/1629110434867
本文版权归作者和AIQ共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出