草稿 ES数据同步方案_max(update_time)

1、中间件取名

hd-sync

---

2、架构图

请点击：架构图链接

---

3、模块说明

如架构图所示，hd-sync具备如下模块：

• 业务数据（业务方）
• 数据抽取
• 数据通道（kafka）
• ES-Adapter
• 查补模块
• Elasticsearch

下面将一一介绍。

---

3.1 业务数据

概述：该模块并非hd-sync所有，而是说明一下，如果业务方需要接入hd-sync，需要做哪些约定。

单表抽取

• hd-sync为通用且与业务分离的，不可以掺杂与业务强耦合的复杂SQL语句（如Join语句）。
• 如果业务方的数据来自于多张表，业务方需自行创建中间表，融合join多表之后的结果，将该中间表作为数据源，提供给hd-sync。
• 如果业务方对创建中间表的改造难度巨大，则可以考虑将join语句组建成view（视图）。但由于性能会下降，架构组并不推荐此形式。
• 如果业务方只能提供多表关联的 join sql语句，则应将主表中的update_time作为增量标识，且要为主表建立别名。

业务唯一标识

业务方提供的数据源（sql语句），必须有全局唯一业务标识（字段），如订单号，运单号，发票号。

自增主键id

业务方提供的数据源（sql语句），必须有自增id作为自然主键。

update_time

业务方提供的数据源（sql语句）中，针对update_time字段，有如下要求：

1. 必须要有且只有一个，能权威代表数据的更新时间；
2. 该字段必须为索引（且为独立索引）；
3. 该字段必须严格体现数据更新的时间戳：数据更新，则update_time必须更新；

联合索引 id+update_time

为实现在update_time范围跨度较大时仍能做到索引覆盖，业务方的数据表应该提供联合索引(id,update_time)。对应sql为：

select id from t where id > x AND id <= periodMaxId AND update_time between m and n order by id limit pagesize;

---

3.2 数据抽取

概述：该模块的职责在于数据抽取，衔接业务方数据源与数据通道（kafka），并不直接与ES交互，而是将写入ES的工作交给了后续的ES-Adapter来完成。
该模块包括：手动增量，手动全量，定时增量，三个子模块。下面将一一介绍。

3.2.1 方案选型

HA方案选型

结论： 我们选择“db表+守护线程”的方案，没有必要只为了一个集群节点的检测，而引入一个zk架构。

增量方案选型

结论： 我们选择“自研定时增量”，可以应对更多个性化的业务问题；如果追求极致实时性，可采用业务方双写的形式（ES-Adapter会提供用于业务方双写的开放API）。

3.2.2 数据库表设计

数据源信息

表名：offset_info

集群节点

表名：node_info

增量批次

表名：batch_info

3.2.3 线程设计

增量数据抽取任务分配线程

1. 每个节点内部都会有一个数据抽取的任务分配线程。该线程会遍历当前节点负责的所有表，对每张表创建一个抽取任务。抽取任务的关键要素：表，抽取时间段（开始时间，截止时间）。每构建一个抽取任务，都提交到线程池执行。
2. 任务分配线程构建任务的关键流程。该线程在分配任务前，会对每张表查询其max(update_time)是否大于当前已经记录的last_offset。如果大于，则创建任务，提交给线程池。任务参数为：表名，last_offset，max(update_time)。
3. 任务分配线程的执行频率。在准备根据某张业务表来构建一个新的任务前，需要先校验该表是否有数据抽取任务在健康执行。具体校验流程为：查询batch_info表中，是否有当前业务表正在RUNNING/WAITING的批次，并校验其update_time心跳值是否大于设定阈值，如果均合理，则暂不构建任务；否则，则构建任务。【说明：针对RUNNGING/WAITING状态的批次，如果其心跳值已经超时，即任务线程崩溃，则需要根据success_count/total_count的比率，来判断该批次应该是重新构建任务、或者交由插补线程后期处理。】

数据抽取执行线程

寄生在线程池中，根据任务参数（表名，开始时间，截止时间）来执行具体的数据抽取。开始执行任务时，需要更新batch_info表的status为RUNNING；在任务执行过程中，需要每处理X条数据，就更新一次batch_info表（的success_count，fail_count，update_time字段）【说明：X是可配置的，要控制在合理的心跳超时阈值范围内，如果处理X条数据一般需要10秒，则心跳超时阈值不能低于10秒】；全部执行完成之后，需要更新batch_info的status字段为COMPLETED。

集群节点心跳线程

1. 每个节点都会有一个后台线程，每X秒做一次心跳。心跳，即为更新node_info表中当前节点记录的heartbeat_timestamp字段为当前时间。
2. 如果node_info表中没有当前节点的信息，则insert一条记录，heartbeat_timestamp字段为当前时间，并触发重平衡。

集群节点健康检查线程

1. 每个节点都会启动一个后台线程，每隔X秒，遍历node_info表记录，检查heartbeat_timestamp字段与当前时间的距离值，如果大于阈值，则物理删除该记录。
2. 每次遍历时，首先检查node_info表中第一条记录（即最小id记录）的心跳字段是否超时，同时检查其是否为当前节点。
   2.1. 如果心跳未超时且为当前节点，则继续执行健康检查逻辑（步骤1）；
   2.2. 如果心跳未超时且不为当前节点，则放弃本次执行，等待下次定时任务；
   2.3. 如果心跳超时且不为当前节点，则以同样的检查逻辑，校验下一条记录。
   以此保证：所有节点中，只有id最小的健康节点，会负责所有节点的健康检查。
3. 无效节点删除，会触发重平衡。

3.2.4 流程设计

手动全量开启定时增量

手动全量往往是在新表初始化上线时，需要触发一次；在手动全量彻底结束之前，不可以提前开始定时增量任务；所以，应该在手动全量开始时，记录偏移量，在手动全量完成之后，将该偏移量入库，此时，定时增量才可以根据该偏移量，开始执行；

任务分配时记录增量批次

在任务分配线程把增量任务提交到线程池之后，就应该记录批次入库，状态可以为WAITING。不能等待任务真正执行时再记录。

增量抽取数据为分段抽取

1. 增量抽取为“分段”抽取，防止因为大量数据被刷导致的增量数据过多。首先统计本次抽取的时间段内，总行数是否大于阈值L（如10万）。
1.1. 如果大于L，则将该时间段内的所有数据的id全量查出放于list（1000万个id大约占用内存70M，可进一步确认），然后每L条数据一组（首先要确认一下，10万条完整数据要占用多大内存），做in语句查询（in语句不能大于4M）。把每组的查询结果，放于kafka中。
1.2. 如果小于L，则直接全量取出，然后一次性放于kafka；
2. 数据抽取，应做好日志打印，能通过日志辨识出数据抽取的进度，防止假死、死锁等问题导致跑批记录长期状态为RUNNING，影响下次跑批。

增量抽取为分页抽取

1. 取最近更新时间max_update_time，应该走索引(update_time)。select max(update_time) as max_update_time from 主表;
2. 取时间范围内的id最小值，应该走索引(update_time)。select min(id) as min_id from 主表 where update_time between last_offset and max_update_time;
3. 串行获取当前页的id集合，由数据库引擎自行选择索引（id 或 update_time)。
   select id from t where id >= min_id AND update_time between last_offset and max_update_time order by id limit page_size;
4. 根据当前页的id集合拼装id列表，构建in语句，追加在业务sql后，应该走主键索引。select * from join_sql where orign_where = 'abc' and id in (id集合)。
5. 如果当前页数据量小于pagesize ，停止循环。否则，找到出当前页数据的最大id，设为m，min_id = m+1，重复步骤3、4、5。

全量同步视为增量同步的特殊情况

数据量大而已。

动态创建数据源

业务数据源可能是变化的，如新接入的业务表，可能与当前已接入的表，不是同一个数据库。
我们考虑采用配置的形式，当新接入业务表时，可通过读取新增的数据源配置，自动创建数据源（连接池）。

分表处理

• 业务表可能有分表的情况，如按时间分表，按hash分表，等等。hd-sync将分表视为独立表，如order-202006，order-202007，两张按月分的表，将按照两张毫无关系的独立表来看待。
• 所有分表，需提前创建好；若为动态创建，则在创建时，需告知架构组增加表配置。

动态创建ES索引

在新增业务表时，需做配置。hd-sync可通过读取配置，动态根据字段映射，创建出ES的index（mapping），传给kafka，有ES-Adapter创建ES索引。

业务表变动

增加业务表，删除业务表，均会触发重平衡。

重平衡分配

节点加入

如果有新节点加入，则会在心跳线程的第一次心跳时，执行重平衡。

1. 检查offset_info表中是否有当前节点的记录，如果有，说明已经被其他节点的重平衡线程分配过了，无需再执行；如果offset_info表中没有当前节点记录，则触发重平衡，执行以下步骤。
2. 获取重平衡分布式锁；二次校验步骤1，如果当前节点依然没有被分配，则执行步骤3；
3. 获取offset_info表数据按id排序，存入数组ao，假设有m个；获取node_info表数据按id排序，存入数组an，假设有n个；遍历ao下标，将ao[i]的node_id更新为an[i mod n]。

节点删除

如果有节点宕机，则会在健康检查线程中，待所有宕机的节点都被物理删除之后，执行重平衡。

1. 检查offset_info表中是否有刚刚删除的节点对应的记录，如果没有，说明已经被其他节点的重平衡线程分配过了，无需再执行；如果offset_info表中有刚刚删除的节点对应的记录，则触发重平衡，执行以下步骤。
2. 获取重平衡分布式锁；二次校验步骤1，如果offset_info表中依然有刚刚删除的节点对应的记录，则执行步骤3；
3. 获取offset_info表数据按id排序，存入数组ao，假设有m个；获取node_info表数据按id排序，存入数组an，假设有n个；遍历ao下标，将ao[i]的node_id更新为an[i mod n]。

3.2.5 故障处理

增量同步要互斥

场景描述： 表t目前是有节点node1负责的，并且此时此刻，节点node1正在抽取表t的数据；然而在此时，节点node2加入集群，表t恰好被分配给了node2，此时node2会立即进行数据抽取，导致数据抽取被重复处理。
解决方案： nodeX在做数据抽取之前，会先检查batch_info表，根据schema+table+last_offset+status做重复校验，如果status为RUNNING，则说明其他节点正在抽取该批数据，需等待其为COMPLETED之后再行执行。同时，也要检查batch_info表中node_id对应的节点是否健康（node_info表中是否存在），如果不健康，则立即执行跑批。

---

3.3 查补模块

概述：查补逻辑转为查缺补漏而设。虽然我们已经（在ES-Adapter中）做了失败处理，但我们担心其并不能做到天衣无缝，我们担心依然会有漏网之鱼导致的db与es不一致。这种漏网之鱼，可能是由于数据抽取层漏抓数据，也可能是由于kafka丢失数据，也可能是由于ES-Adapter的未知异常导致。所以，我们增设了这么一个查补模块，上接业务数据源，下接ES。常年全表扫描业务表，并与ES中的数据做校验。

流程设计

业务表与ES数据校验

定时执行，频率暂定每天一次（可配置），预留手动触发按钮。
执行逻辑：

1. 从数据表中筛选出update_time大于上次查缺补漏偏移量（时间戳）的记录。
2. 遍历对比每条记录的id和update_time：如果es中存在对应id的记录且es中对应记录的update_time较大，则丢弃；否则，在业务表中查询出完整数据，并放入kafka相应index的topic中。

失败处理Topic的消费者

在ES-Adapter中，如果写入ES失败，可能会将失败记录与失败原因，发送至kafka相关topic。此时，查补模块将作为该topic的消费者，订阅将采用正则表达的方式es-error-*，订阅所有index的写入失败数据。
目前考虑的处理形式为，将失败数据取出，并写入db中，等待人工处理。

---

3.4 数据通道（Kafka）

概述：该模块解耦“数据抽取”与“数据写入”，将两部分容易产生瓶颈的模块，分隔开来，可以做到单点优化，任何一个模块产生瓶颈，不会影响另一个正常运行。同时也让职责更加明确，如：数据写入对硬件资源的占用，不会干扰数据抽取。

3.4.1 Topic规则

1. 每个ES index，对应一个Topic
2. 所有正常Topic的命名规范为：“es-sync-”前缀 + “index索引名”。如es-sync-order，表示订单。
3. 增设失败处理Topic，用于接收写入ES失败的数据及失败原因。命名规范为：“es-error-”前缀 + “index索引名”。如es-error-order，表示写入订单index失败的数据。
4. 每个分区一次性设置足够的分区数（如12个），以便后续的消费者能灵活横向扩展；

3.4.2 消息格式

统一使用json格式发送和接收消息。

---

3.5 ES-Adapter

概述：ES-Adapter，是衔接kafka与ES的中间层。该模块不参与DB操作，只与kafka和ES有交互：消费kafka中的数据，并将数据写入ES；收集写入ES失败的数据，发回kafka错误主题。

3.5.1 消息订阅

1. 消费者以正则表达式“es-sync-*”来订阅topic。
2. 应该采用多节点消费+多线程处理的形式加速ES写入。【每个集群节点设定一个消费者，每个消费者对应一个线程池】

3.5.2 双写开放API

提供接口，入参为（app_id，index_name，json_data），为业务系统提供双写能力。该接口会直接把数据送至ES，不经过kafka等中间件的流转过程，实时性较高。
意义在于：如果业务系统对数据实时性追求极致，则应该主动采用双写策略，即在写完db之后，应该手动调用该开放API实现双写到ES（不用等待定时增量同步）。这也是追求极致实时性所必须产生的代码侵入性。

3.5.3 查询开放API

提供接口，入参为（索引，查询条件列表，排序信息，分页信息），实现多条件经过组合查询，可以按照任意字段排序。

3.5.4 校验流程

数据一致性校验

在将消费的数据写入到ES之前，会根据ID+update_time先行校验，看ES中的数据与当前消费的数据是否一致；如果ES中无次数据，则新增；如果ES中的数据交旧，则更新；如果ES中数据较新或与当前消费数据一致，则无需任何操作；

字段一致性校验

针对kafka中消费的数据，字段如果与ES中index的字段不一致：针对新增情况，则要触发ES字段新增，针对其他情况，需要重建index；
注意：ES新增字段后，历史数据的查询，应该不能按新增的字段查询出来。

3.5.4 异常处理

针对写入ES失败的情况，方案有二：

1. 将失败数据与失败原因，封装成message，发送给kafka中相应索引的错误接收topic，由该topic的消费者，执行具体的处理逻辑。
2. 打印日志，发出警告（邮件，钉钉，短信，等）。

梯度重试：

• 无论上述哪一种方案，都应该在执行上述流程之前，采取梯度重试，过滤掉那些可以通过重试来解决的失败（如网络抖动）。
• 所谓梯度重试，是指遇到失败时，应分别等待1s，5s，10s，再重试，重试次数与重试梯度，可以通过配置来指定。