大数据开发平台-数据同步服务
什么是数据同步服务？顾名思义，就是在不同的系统之间同步数据。根据具体业务目的和应用场景的不同，各种数据同步服务框架的功能侧重点往往不尽相同，因而大家也会用各种大同小异的名称来称呼这类服务，比如数据传输服务，数据采集服务，数据交换服务等等
至于大数据开发平台的数据同步服务，加上了限定词，那当然是进一步把业务的范围限定在了和数据平台业务相关的一些组件和应用场景之下了。
大数据平台数据同步服务业务场景
讨论场景之前，先来看一下数据同步的目的，为什么我们需要在不同的系统之间进行数据的同步？
从大数据开发平台的角度来说，很显然，是因为我们通常不能直接对线上业务系统所存储或生成的数据进行各种运算或检索处理，组件技术架构是一方面原因，业务安全性隔离是另一方面原因。
所以，我们就需要把这些数据采集到开发平台的各种存储计算组件中来进行加工处理，这个过程也就是所谓的ETL过程。
然后，在开发平台中处理完毕的数据，有时候也并不能或着不适合在大数据开发平台的相关服务中直接使用，需要反馈回线上的业务系统中，这个过程我们称为数据的回写或导出。
最后，即使在大数据开发平台自身的各种存储／计算／查询服务组件之间，因为架构方案，读写方式，业务需求的不同，也可能存在数据的传输同步需求。
从上述三类应用场景来看，我们可以看到，通常来说我们所说的大数据开发平台环境下的数据同步服务，主要处理的是不同系统组件之间的数据导入导出工作。 比如将DB的数据采集到Hive中来，将Hive中的数据导出给HBase之类。也就是输入和输出的数据源是异构的，数据同步的目的是让数据可以适合业务需求的形式，在不同的系统中用各自擅长的方式运转起来。
除此之外，还有另外一种出于数据备份，或者负载均衡的目的而存在的数据同步场景。比如DB的主从同步，HBase集群的Replicator备份等等，他们的输入输出数据源往往是同构的。这类场景下，具体的同步方案和流程通常和系统自身的健康，功能逻辑，服务诉求等有着较强的关联性，所以往往对应的系统会自带同步方案实现，属于系统自身功能实现的一部分，比如MySQL的binlog主从同步复制机制。这类特定系统自带的数据同步架构方案实现，不在本文讨论的范围之类。
数据源
业务范围明确了，那么让我们来看看在这种业务场景下，需要处理的数据源可能都有哪些，简单分以一下类，常见的数据源大致可以分为；
- 关系型数据库类 : 比如 MySql, Oracle, SqlServer, PostgreSQL 等等
- 文件类：比如日志log，csv，excel等各种传统单机文件
- 消息队列类：比如kafka和各种MQ
- 各种大数据相关组件：比如HDFS／Hive／HBase／ES ／Cansandra
- 其它网络接口或服务类：比如FTP／HTTP／Socket 等
现有的解决方案介绍
如上所述，数据同步服务可能涉及到的外部系统多种多样，实际上，但凡能存储或产生数据的系统，都可能成为需要接入数据同步服务的数据源。因此，也不难想象，市面上一定存在众多的解决方案。
为什么呢？很显然，这些各式各样的数据同步服务方案，在不同的业务场景中，无论整体功能定位还是业务覆盖范围都可能千差万别。即使某些方案的业务定位类似，在具体的功能实现方面，大家关注的重点也可能有所区别。此外部分系统在设计的时候，为了保证易用性或者提供一站式的解决方案，其架构和具体的功能逻辑与上下游系统可能还有一定的业务关联性，再加上程序员又喜欢用各种开发语言来折腾一遍, Python／Java／Ruby／Go 。所以，这类服务系统的解决方案想不多也很难啊。
那么常见的解决方案都有哪些呢？
以关系型数据库为主要处理对象的系统：
Tungsten-replicator
是Continuent公司开发的数据库运维管理的两个相关联的工具产品套件中的一个，负责Oracle 和 MySQL两个数据库之间异构的数据复制同步工作，以及对外导出到比如Redshift，Vertica等数据库，也包括导出到Hadoop环境。
Continuent的另一个产品是tungsten-clustering，为MySQL等数据库提供拓扑逻辑管理，灾备，数据恢复，高可用等功能，很显然这些功能很大程度上是和Replicator相结合的。
我们没有真正使用过Tungsten的产品，只是在架构和代码方面有过一些调研了解，总体感觉，作为商业解决方案，架构完善，但是相对复杂，一些业务流程方案是定制化的，对关系型数据库自身的数据同步和管理以及稳定性应该是它的强项，不过作为一个开放的系统来用的话，接入成本可能有点高。
Canal 和 Otter
Canal是阿里的MySql增量数据同步工具，Otter则是构建在Canal之上的数据库远程同步管理工具。两者结合起来，产品的目标范畴大致和Tungsten-replicator差不多。
和Tungsten类似，Canal也是基于MySql的Binlog机制来获取增量数据的，通过伪装MySql的Slave，来获取Binlog并解析增量数据。
这两种方案通常是大家用来接入Mysql binlog的最常见的选择，毕竟MySql Binlog的格式解析模块也是一个相对专业化的格式逆向工程工作，即使不直接使用这两个方案，大家也会借用这两个方案中的binlog Parser模块来做二次开发。
Canal的主要优点是结构流程比较简单，部署起来并不太难，额外做一些配置管理方面的改造就可以更加自动化的使用起来。不过，主要问题是Server和Client之间是一对一的消费。不太适用于需要多消费和数据分发的场景。
我司之前既有对Canal简单的封装应用，也有在借用Canal的Binlog Parser的基础上，开发的DB增量数据分发系统
阿里的DRC／精卫等
DRC按阿里官方的说法定位于支持异构数据库实时同步，数据记录变更订阅服务。为跨域实时同步、实时增量分发、异地双活、分布式数据库等场景提供产品级的解决方案。
其实精卫也是类似的产品，不过是由不同的团队开发的，除了这两者，阿里内部可能也还有过其它大大小小类似的产品，最后大概都整合合并了，DRC胜出 ；）
从定位说明就可以看到，很明显，除了点对点同步，DRC还需要支持一对多的消费和灵活的消费链路串联，对性能，顺序一致性等方面的要求也可能会因此而变得更加复杂（未必更难实现或要求更苛刻，但是可能有更多不同角度的功能需求），比如，可能需要支持有限时间段内的回溯，和精确定位消费的能力等。
DRC相关系统，阿里并没有开源，不过我司之前和阿里的同学有过一些简单交流，我们也从中了解和学习了一些产品设计和架构方面的思想。
我司的DB增量数据分发系统Pigeon的第一版，就是在Canal的Binlog解析代码模块基础上，参照DRC的部分思想进行开发的。大致的方案是将前端的Parser对接到消息队列上，让消息队列来承担消息持久化和分发的工作，此外在Server层面再辅助以服务节点和消费链路的动态管理，负载均衡，加上数据的过滤，转换和分发模块及策略的管理，消费端SDK的封装等工作。
以日志或消息队列为主要业务对象的系统
这类系统一开始可能是以日志查询为主要业务场景，其中的数据同步服务相关组件，有些是独立的组件，有些则是一整套采集，计算，展示等完整的业务方案中的一部分。不过随着架构的不断发展和成熟，有些系统渐渐的也不仅仅是只定位于处理日志类业务场景了，开始向通用数据采集传输服务的角色靠近。
Flume
Flume现在大家用的多的，是Flume-NG这个redesign过的版本，它的定位是离线的日志采集，聚合和传输。Flume的特点是在聚合传输这块花了比较多的力气，特别是早期的版本，需要配置各种节点角色，而在NG版本的设计中，其拓扑逻辑已经简化成只有Agent一个单一角色了
通过Agent的串联可以构建出复杂的数据传输链路，此外还有事务的机制设计来确保链路传输的可靠性。 不过，个人觉得，由于Kafka等通用消息队列的广泛使用，Flume在聚合，传输这方面的作用，在一些场景下其实是可以通过其它方式来实现和弱化的（比如没有网络带宽或远程二次传输问题的场景中）。
我司的日志链路中，也没有使用Flume，而是采用了自主研发的Agent采集器直接对接Kafka（当然，一些场景下，也未必绝对合理）
LogStash
LogStash是著名的ELK套件中的一个组件，负责日志采集和转换，另外ES负责存储和检索，Kibana负责查询结果的展示
LogStash从设计上来看，在数据传输和链路串联方面的考量就简单了很多，它的重点放在了数据的转换处理上。所以它在过滤器，编解码器等环节下了很多的功夫，比如支持grok脚本做过滤器逻辑开发，在内部链路上还有各种的buffer设计，用来支持数据的合并，转换，条件触发输出等功能
应该说从数据转换处理的角度来看，个人觉得LogStash的设计已经足够灵活和完备了。不过，它的主体实现语言是Ruby...
Camus
Camus严格的说算不上是一个框架，它是Linkedin开发的基于Kafka消费日志，批量写入Hdfs的一个工具，不过用的人也不少，所以提一下，我司之前也有大量的日志是通过Camus来采集的。（话说Linkedin是把自家的kafka用到极致了，各种链路但凡能依托kafka实现的，大概都不会考虑其它的实现方式）
Camus的架构方案，基本上就是写了一个MR任务，实现批量从Kafka读取日志并写入Hdfs，此外自身维护了kafka中各个topic的消费进度。用它来做kafka topic的简单映射采集，稍微做一点管理方面的适配开发，基本还是可行的。不过，它的缺点主要是对Topic进行定制化的处理比较困难，需然也提供了一些Hook接口，但是毕竟架构过于简单，对数据进行一些Per topic的过滤转换工作，就有点力不从心了。
另外一些想不出怎么强行分类的数据同步解决方案 ；）
Sqoop :
Sqoop大家应该不陌生了，即使没用过总应该也听过，也有不少公司使用Sqoop来构建自己的大数据平台数据采集同步方案。Sqoop从一开始，就几乎是完全定位于大数据平台的数据采集业务的，整体框架以Hadoop为核心，包括任务的分布执行这些，多半都是依托MR任务来实现的。数据同步的工作，也是以任务的方式提交给Server来执行，以服务的形式对外提供业务支持。
Sqoop的处理流程，定制化程度比较高，主要通过参数配置的方式来调整组件行为，在用户自定义逻辑和业务链路流程方面能力比较弱，另外，依托于MR的任务处理方式，在功能拓展方面也有一些约束和局限性。此外各种数据源的输入输出实现部分，稳定性和工程实现细节方面，也只是可用，但算不上完善和成熟。
我司也没有使用Sqoop来构建大数据平台的数据采集，导入导出服务。上述原因虽然是一方面原因，但绝对不是主要的原因。最主要原因还是因为数据的采集和导入导出服务体系，具体的输入输出模块的构建只是一部分内容。更重要的是要构建任务的配置，管理，监控，调度等服务，以及对整个数据同步业务流程和生命周期的封装，和对用户交互体验及产品形态的完善。理想中，需要和开发平台整体开发环境深度集成。
DataX
DataX是阿里开源的一款插件式的，以通用的异构数据交换为目标的产品。其核心思想，简单的说，就是之前阿里的同学写各种数据源之间的同步工具，都是点对点的实现，写多了以后，发现这种两两之间网状链路的开发代价比较高。而DataX呢，是通过标准化的输入输出模块，将点对点的实现变成了星形的拓扑结构，增加一个数据源只要单独写这个数据源的输入输出实现模块就好了。
其实，这个思路也没什么大不了的，和前面的Flume／LogStash等的思想并没有本质的差别，人家一开始就没有走网状结构的路 ：）
不过，DataX的特点是内部的结构更加简单一些，没有channel啊之类的概念，不具备持久化能力，也没打算构建复杂的数据流动链路。你可以认为它本质上就是将两个数据源之间点对点的传输工作模块化标准化了，最终构建出来的，还是一个简单的进程内直连的数据读写链路。
此外从一开始，DataX的目标就是在简化新链路开发代价的基础上，追求数据的传输效率。比如，使用了Ringbuffer类的技术来做input和output模块之间的数据转发工作。
因为DataX简单和标准化的特点，所以也有不少公司基于DataX来构建自己的异构数据交换服务系统。
当然，DataX也在持续改进中，目前的3.0版本在作业的分片处理，业务容错，数据转换，流量控制等方面也做了不少的功能拓展。
Heka
Heka是Mozilla开源的一套流式数据采集和分析工具，最主要的架构实现，其实也就是数据采集同步这部分框架。整体的结构设计和LogStash等系统看起来大同小异。这个系统，我并没有做过实际的实践应用，只是简单了解了一下产品设计，提它呢，是因为架构看起来也相对比较完善，另外，它是用Go写的，偏底层后端服务开发的同学可能会喜欢。
数据交换服务产品设计和需求分析
从前面的业务场景讨论和市面上常见的系统的介绍中，你应该不难看出，数据同步是一个业务覆盖范围很广的术语，具体的产品形态设计和功能需求，其实在很大程度上取决于你所定位的业务的职能范围。
我司的大数据开发平台中，数据交换服务系统的定位，和DataX比较类似，系统的功能和产品形态定位，是异构数据源之间的点对点数据读写链路的构建。至于比如业务端的数据采集，数据分级传输链路的构建，增量数据的分发，数据库同步拓扑逻辑管理等环节，并不在我们的数据交换服务系统的功能定义范围之内，这些环节并不是不重要，只是在我司的实践中，是由其它的系统来独立提供服务的。
而点对点的数据读写链路服务产品的组成，又可以分为两部分，一是底层具体承载了单个数据交换任务的插件式的数据交换组件，二是上层的数据交换任务管控平台，其职能范围不仅包括系统和任务自身的配置运行管理，有时候还需要考虑针对上下游系统和具体业务的一些特性进行流程上的适配和定制。
下面的讨论基于上述产品定位展开：
数据交换底层组件
底层组件设计需要关注的地方，在前面的各种开源系统的介绍中，其实大多都已经涉及到了。
首先，从框架结构的角度来说
整个数据的读写转换流程，理想中当然是每个环节都能以Plugin的方式进行灵活拓展。链路环节拆得越细，定制能力当然就越好，但是要同时保持系统整体的易用性也就相对更加困难一点。
那么，数据交换读写链路的分解，大致可以分为几个模块呢，往大了拆分，差不多就是：输入，过滤转换，输出这三个模块。
再细化一些，为了提高模块的复用能力，那么还可以从输入模块中拆解出Decoder模块，从输出模块中拆解出Encoder模块
最后，为了达到数据链路复用的目的，还可以在输出模块之前增加一个路由模块，将一份数据拆分或复制输出到多个目标源中。不过，如果在框架中引入了这样的设计，实际上是将业务流程方面的复杂度下沉到底层组件中来，是否值得，如何取舍，就要看整体系统的设计思路了。
其次，从性能的角度考虑
为了提升性能，除了要求执行节点具备水平拓展能力，还需要考虑支持单个作业的分布式执行能力。
前者如何实现取决于数据同步服务系统的架构设计，如果是采用server模式的服务，客户端提交任务请求到服务端执行的，那么需要Server端能够水平拓展任务的worker执行节点，这个通常不会太难，就是需要自己管理工作节点，或者依托其它集群服务，比如提交MR任务到Hadoop集群上执行。而如果采用的是本地进程模式，客户端在哪里发起调用就在哪里执行，那么资源调度和负载均衡的工作，通常就会上移到工作流调度系统上来管理，数据同步服务自身不负责工作节点的管控。
后者，单个作业的分布式执行能力，实现起来就复杂一些了。因为这涉及到单个作业内部数据的分片处理。当数据源是Hadoop类的系统时，由于这类系统从架构设计的角度，天生就支持数据分片的能力，所以实现起来通常都不会太困难，但是对于DB，消息队列类的数据源，如何实现分片，往往就要复杂一些了。
以DB扫表任务为例，你要分片执行，那就需要数据表具备分段检索的能力，最好是可以基于主键索引进行分段检索，否则只是单纯的条件过滤，会大大加大对DB的压力。但是，现实应用中，很可能并不是所有的表都具备确定范围的主键，有些主键也可能是非连续离散的，这些都会导致很难均衡的对数据进行分片，进而影响分片执行的效率。
另外，基于DataX这种输入输出端独立插件思想构建的数据交换链路，如何和Hadoop体系的数据源的数据分片处理流程更好的结合，充分利用好原生的分布式处理能力，也是需要仔细构思的。
最后，从业务稳定性的角度考虑
要保证业务的稳定性，从底层组件的角度来说，整体系统的流控和失败重试这两个环节往往也是需要重点考虑的。因为数据交换服务所对接的外部存储系统，通常还承载了其它的业务。所以其负载能力往往都有一定的约束要求，其业务环境也不是完全可控的。因此数据交换服务组件，需要能够约束自己的行为，同时应对可能发生的错误。其目的，都是为了提升整体链路的稳定性，降低维护代价。
数据交换服务管控平台
作为服务，不提供可视化的管控平台，只提供命令行交互方式，那就是耍流氓。
管控平台管什么？首先，当然是管理数据交换作业的任务配置信息了
标准的做法，基本都是让用户通过UI界面，以参数的形式配置任务信息，比如输入输出数据源，表格，字段信息，分区信息，过滤条件，异常数据处理方式，调度时间，并发度控制，流量控制，增量或全量配置，生命周期等等。总之，就是尽量让用户能够通过配置信息来表达自己的业务诉求。
当然，任务可供配置的参数越多，使用起来可能就越繁琐，此外，一些复杂的过滤，聚合或转换逻辑，很可能也没办法简单的用配置的方式进行表达，这时候就需要考虑提供自定义组件的管理方式了。
除去数据交换任务自身配置信息的管理，数据交换服务管控平台需要提供的其它服务，其实和大数据开发平台上其它类型的作业任务的管理十分类似，比如:
- 提供数据交换任务的执行流水信息，便于用户查询任务执行情况和进行业务健康分析
- 提供权限管控和业务分组管理，更好的支持多租户环境应用场景
- 提供系统流量负载监控，任务错误跟踪报警等，更好的支持日常的系统及业务运行维护工作。
这些服务可以由数据交换服务平台独立提供，但最理想的，还是和开发平台的其它作业任务融合到同一个平台上进行管理，即使底层支撑对应服务的后台可能是独立的，在用户交互后台上，也要尽可能集成到一起。一方面减少重复开发的代价，另一方面，降低用户的学习使用成本。
上下游系统和业务流程适配
你无法左右别人，但是你可以改变自己。很多时候，数据同步服务，需要配合上下游系统，进行必要的流程定制，来满足业务的需求。
数据结构变更
数据同步业务，最经常遇到的问题，就是业务DB的数据结构发生变更，导致任务运行失败。
数据结构的变更，通常很难自动解决。比如用户自定义了数据扫描的语法，当数据结构变更以后，已经非法了；比如源表的字段信息发生了增删改，目标表如何映射适配？历史数据能否转换处理，是否需要转换处理？另外，不同的数据源，增删改的处理方式也可能不同，业务方希望采取的应对方式可能也和具体业务逻辑相关。所以，很多情况下，数据结构的变更，都是需要人工干预的。
那么系统能做些什么呢？自然是通过工具尽可能的降低这个变更过程的代价，比如
- 监控源表元数据的变更，提早发现问题，提早解决，避免在半夜真正执行任务时才出错报警
- 规范业务流程，比如约定字段的变更方式，变更的通知机制等，通过最佳实践降低问题风险概率
- 对一些已知场景提供标准化的自动处理方式，减少人工干预的需要，加快数据转换，重建处理流程等等
数据时间问题
在离线业务中，大量的数据导入任务都是在凌晨附近导入前一天的数据进行批处理分析。这种场景下经常可能会遇到以下问题：
- 数据可能由于各种原因晚到，在数据导入任务开始执行的时候，前一天的数据还没有完全到位。
数据晚到的可能原因很多，比如DB主从延迟太大，客户端上报不及时，业务端采集链路因为流量或负载或故障等原因未能及时采集数据等等。
这时候，通常的做法，一是将日常数据采集时间适当往后推迟一小段时间（比如15分钟到半个小时）降低问题出现的概率。二是往往需要对各种链路已知可能延迟的环节进行监控，比如采集DB主从延迟时间，队列消费进度等等，及时报警或阻断下游任务的执行。三是对晚到的数据，需要根据业务需求制定适当处理策略，是丢弃还是补充回写到前一天的数据中，还是直接划入第二天的数据里等等。
- 数据本身没有手段区分业务更新时间，具体执行结果依赖于任务执行时间
比如DB扫表的任务，如果表格中没有用于区分业务时间的字段，但是统计业务中却需要按日期划分统计，就只能靠凌晨精确的时间点采集来实现了，这就很尴尬了，因为你很可能无法保证任务开始执行的时间。你可能会说这种情况是DB表结构设计得有问题，的确如此，这时候就需要推动业务方进行改造了。
还有一种情况更常见一些，就是DB表格中的确存在业务更新字段，但是，同一主键的数据可能有多个状态变迁，会被更新多次。而时间戳只有一个。举个例子，比如你有一个订单信息表格，里面记录了下单，付款，发货，收货，确认等等不同的状态，但是，只有一个update字段。那么根据某一个时间点扫描的数据，你可能无法判断出这些状态发生变化的准确时间，那么就有可能发生统计归属错误或者遗漏的情况。
这两种情况，通常都是因为业务方的业务流程本身并不依赖于这些时间信息的记录，但是做数据统计的时候需要这些信息，而业务开发方和数据统计方负责的同学是两拨人，开发方没有充分考虑统计的需求。
有时候这种情况问题也不大，比如半夜业务变更不频繁，数据采集过程迟一些早一些，数据偏差都不大，或者这类数据统计到前一天还是后一天都没有太大的关系。但是，当出现大范围时间偏移，或者你需要重跑历史数据的时候，比如今天重跑上周的数据，那么从当前DB快照无法复原业务字段变更的具体时间点，就会成为一个无法忽视的问题了。
总体来说，这类问题的解决，首先数据同步服务自身得提供根据业务时间过滤数据的手段，其次要推动业务方改造数据结构，避免出现无法还原的场景，最后，有些业务还可以通过采集binlog等实时增量的形式，通过分析每次数据的具体变化时序来解决（当然，由于log保存时间有限，对于长时间跨度重跑的场景，是无法通过这种方式来解决的）
分库分表处理
分库分表，大概是业务上了规模以后，大家都喜欢做的事。但是DB中分表可以，导入到比如Hive中以后，你得想办法合并啊，便于后续各种运算逻辑的开发和统计查询脚本的撰写，那么问题来了：
比如你是通过扫表的方式获取数据，如果没有类似阿里的TDDL这样的分库分表中间件来屏蔽DB分库分表细节，你会需要自己处理相关逻辑，管理和连接所有数据实例。如果走binlog获取数据，在分库的场景下也需要自己想办法合并数据采集流程和结果。
更麻烦的是，如果你的业务方分表设计的时候，不够规范，不同的分表之间没有唯一的主键可以加以区分（可以区分的字段，也可能不是主键），那么在合并数据的时候，你可能就需要允许用户自定义合并用字段，或者自动捏造出一个主键出来，避免数据的冲突
这个问题同样，最理想的解决方案也是通过推动DB分库分表中间件的建设和业务规范的建立来解决，但这对很多公司来说往往不是一件简单的事，所以，在此之前，就需要自己想办法定制解决了。
数据合并去重等
通过binlog增量方式来获取DB变更数据，优势是时效性好，有时也是某些场景下唯一的解决方案。但是因为走Binlog来给离线批处理任务同步数据，实际上，数据是经过了表-流-表这种模式的切换，而这种切换也会带来附加的问题
从表的变化解析成数据流，这个过程问题不大，但从数据流重新构建回表格，就会有几个问题需要关注了：
- 取决于数据流传输的方式，数据流可能发生乱序，重复的问题，对重构表格带来困难。
比如用消息队列传输数据，各个分区的数据可能无法保持全局有序性，消息队列本身可能也无法保证Exactly once的投递。如果业务流程不能允许这类问题的发生，那就需要针对性的加以防范了，比如结合业务知识，使用合适的分区字段，使得局部有序的数据对业务结果不会造成影响。
- 目标端数据源，比如像HDFS或Hive文件，可能只允许添加记录，或全局重写，而无法单条删除或者更新记录。
这种情况下如果源端数据源类似DB中，一条记录发生多次变更，就会生成多条变更记录，而下游任务比如一天的批处理任务，只需要最后凌晨时间点上的状态信息，这时候就需要对变更记录进行合并了。
合并数据的可能方法很多，取决于具体的业务场景和代价，未必有统一的最佳方案。首先你需要解决数据乱序问题，然后：
你可以在数据流式采集方案的后端，将数据先写入一个支持单条记录删改操作的中间数据源，然后到点再从这个中间数据源导出最终数据到目标数据源。
如果数据量不大，你也可以在采集程序中汇总所有数据，去重完再写出到目标数据源。
你也可以不去重直接将所有变更流水写入目标数据源，事后再运行一个清理程序进行去重，前提是除去采集时间，原始数据中还具备可以用作去重判断的依据。
我司相关服务的现状和未来改进计划
目前我司的数据交换服务，日志相关链路，采用Camus和自定义的Hive Kafka Handler两种方式采集，后者在采集的基础上添加了Per topic的过滤转换逻辑，可以通过自定义Hiveql一步完成数据的采集和转换工作。
其它大数据组件之间以及与DB间的数据交换服务，由自研的与DataX类似架构的系统承担，插件式开发，能够处理增量／全量，并发流控，分库分表等前面所描述的常见需求。另外，管控平台基本实现了用户可视化的配置，管理，执行流水查询，变更记录查询，系统负载和业务进度监控报警等功能。此外，在数据交换任务的数据质量监控方面，也做了部分采集和统计分析工作。
整体来说，主要的服务框架流程没有很大的问题，但是在与开发平台的整体集成和用户自助服务的易用性方面与理想的状态还有很大的差距，其次在性能，稳定性，拓展性等方面也有很多工作等待开展，在数据质量监控方面做的工作也相对粗糙，所以未来的改进方向，包括：
- 底层数据交换组件的进一步模块化，标准化，重点加强用户自定义数据过滤和转换模块的建设
- 单个作业分布式分片处理方案的改进，提升大表同步作业的处理效率
- 数据合并／去重方案的改进，提升性能规避容量瓶颈（目前的变更合并工作还是通过二次写入专属DB来实现）
- 任务流量，负载，进度，异常等Metrics信息的全面采集和汇总分析，便于及时发现问题，持续改进业务
- 全链路的分级容错和自动重试恢复机制的完善改进（目前的容错重试机制是作业级别的，粒度太粗）
- 更加自动，更加平滑的流控和负载隔离机
- 数据交换服务管控后台与大数据平台整体开发环境的进一步融合，提升用户自主服务能力，降低业务开发维护成本
- 完善异常，错误反馈机制: 比如对常见问题，汇总，解析后再明确的反馈给用户，可能的话，提供解决意见和方案，而不是直接抛出异常代码，降低用户支持的代价。
- 前述业务数据时间问题的全面推动改进，降低数据同步任务结果的不确定性
小结
总体来说，大数据开发平台的数据同步服务的构建，可以参考的方案很多，具体的读写组件的开发也并不困难，能够找到很多现成的解决方案。对于多数公司的大多数业务来说，底层不论采取什么方案，通常都是可行的。所以数据同步服务建设的成熟度水平，往往体现在管控平台的服务能力水平和业务接入及运维代价的高低。
常按扫描下面的二维码，关注“大数据务虚杂谈”，务虚，我是认真的 ；）
