Streamsets-JDBC模式使用更新时间字段数据同步

StreamSets的开源地址：https://github.com/streamsets/datacollector-oss
Streamsets官网地址：https://streamsets.com/
Streamsets文档地址：https://docs.streamsets.com/portal/datacollector/3.16.x/help/index.html

我又来写Streamsets了，各种原因好久没研究Cassandra了。
本次分享主要介绍Streamsets的JDBC模式、为什么使用时间字段同步数据、遇到的问题和解决方案。
解决方案并不是最完美但也是基于当前条件下最优解，如有疑问，欢迎热烈讨论。
提供的脚本毫无保留，可直接使用。

Streamsets在3.22.2之后就闭源了，更高阶的特性已包装为平台产品。

结合周边讨论和网上的资料来看，Streamsets的活跃度不高，在网上搜的资料太少啦，又随着项目闭源，活跃度更低了，归其原因我分析Streamsets是一个大而全的数据同步工具，整合了市面上基本所有的数据源，但是每个公司不可能用到里面所有的数据源，真正能用到大部分数据源的公司，规模肯定大到不会依赖这种外部的工具，自己手写同步的自由度和效率要更好。

Streamsets对于我们的优势在于开箱即用，相比于手搓代码来实现业务细节，Streamsets将数据同步的每个阶段独立开来，将业务变动最大的数据清洗部分以处理器的形式开放出来，数据的转换和转换的实时配置并生效，直观的监控指标。

版本为Streamsets的3.16.0的离线版本，部署到内网时的最新版本为3.16.0，所以方案和问题的解决方案均以3.16.0为基础。

JDBC模式介绍：

JDBC模式的增量模式只支持新增的数据和不需要修改的数据，且官方建议的offsetColumn为PrimaryKey，如：ID。

Incremental mode
When the JDBC Query Consumer performs an incremental query, it uses the initial offset as the offset value in the first SQL query. As the origin completes processing the results of the first query, it saves the last offset value that it processes. Then it waits the specified query interval before performing a subsequent query.
When the origin performs a subsequent query, it returns data based on the last-saved offset. You can reset the origin to use the initial offset value.
Use incremental mode for append-only tables or when you do not need to capture changes to older rows. By default, JDBC Query Consumer uses incremental mode.

SELECT * FROM <table_name> WHERE <primaryKey> > ${OFFSET} ORDER BY <primaryKey>

这样支持的场景为不断的增量数据，无法捕获数据的更新。
但是正常的业务系统一般不存在只新增不更新的场景。
全量同步模式每次加载所有的数据，当表的数据量较大时，同步所需的时间和延迟不能接受。

修改为通过update_time来捕获数据变化：

SELECT * FROM user WHERE update_time > ${OFFSET} ORDER BY update_time

在配置管道时将OffsetColumn指定为update_time，业务系统使用mybatis-plus在数据新增和更新时补充创建时间和更新时间。数据库的时间精度为秒。

使用update_time的好处是对于开发者和运维人员可读性更好，在进行历史数据的同步和数据对接时更方便。
该方案看似非常合理，业务侧只要控制好update_time的逻辑，每次数据变化时update_time是不断滚动向前的，滚动查询不断的进行数据同步。
但是too young too simple。
按照Streamsets的处理逻辑，在两种场景下会丢数据。
分别是当单次同步的数据量超过maxBatchSize时，概率性丢数据和并发写入数据库时概率性丢数据。
这两种丢数据的场景是不可控的，时间不可控，完全看运气。但是不确定往往是最可怕的。

为什么会丢数据？

第一种场景：单次同步的数据量超过maxBatchSize
Offset的更新逻辑和jdbc-protolib源码中的逻辑：
origin会当根据sql查询的数据读取不超过配置的maxBatchSize的数量，并将最新的update_time赋值给offset。

// com.streamsets.pipeline.stage.origin.jdbc.JdbcSource.java
public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) {// ...try (Connection connection = dataSource.getConnection()) {if (null == resultSet || resultSet.isClosed()) {// 执行查询sql语句resultSet = statement.executeQuery(preparedQuery);}// 超过maxBatchSize的数据不发送到下一阶段，留到下次操作时处理。while (continueReading(rowCount, batchSize) && (haveNext = resultSet.next())) {final Record record = processRow(resultSet, rowCount);if (null != record) {// 记录下数据batchMaker.addRecord(record);}// 更新offsetif (isIncrementalMode) {nextSourceOffset = resultSet.getString(offsetColumn);} else {nextSourceOffset = initialOffset;}// 后续收尾工作}}return nextSourceOffset;}

结合Streamsets的Offset的更新逻辑和jdbc-protolib源码中的逻辑，当一秒内出现多条数据时，会因为精度问题导致数据丢失。

第二种场景：数据并发写入数据库时。
业务侧代码使用mybatis-plus作为ORM来处理数据的读写，当有大数据量写入数据时，如：Excel导入或高并发的数据写入。
mybatis-plus的内置处理逻辑为分批次提交，每次提交1000，所以单个线程写入的qps为1000。
以Excel导入为例，如果批量保存方法没有加@Transaction注解，会大大增加数据丢失的概率。

原因为结合mybatis的处理+没加@Transaction注解导致1000个insert语句一次性发给数据库，这1000条sql语句是以非事务的方式执行，每条数据都是一个完整的事务，执行完毕自动提交，立即可见。
这时当Streamsets触发查询操作时，时机恰好出现在一秒内的前半段，而一秒内的后半段还在数据写入，导致后半段的数据丢失。
场景分析

解决方案：

如果你拿到的是Streamsets的安装包，那第一种场景无法通过配置和升级的方式解决，因为使用的方式和增量模式的设计初衷不符。
有一份折中方案，但不保熟：
1.能力范围内update_time的精度越细越好，越细会有一定的性能损耗，但丢数据的概率大大降低。
2.评估每次同步的数据量大小，maxBatchSize的大小要大于单次同步的数据量。注意内存大小，小心OOM，（插一句：oracle的批量更新会存在连接泄露，需注意。如果有源码顺手改之。）
可以下载一份Streamsets的源码，改之。
代码如下：

// com.streamsets.pipeline.stage.origin.jdbc.JdbcSource.java
public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) {// ...try (Connection connection = dataSource.getConnection()) {if (null == resultSet || resultSet.isClosed()) {// 执行查询sql语句resultSet = statement.executeQuery(preparedQuery);}while ((haveNext = resultSet.next())) {if(continueReading(rowCount, batchSize)){final Record record = processRow(resultSet, rowCount);if (null != record) {// 记录下数据batchMaker.addRecord(record);}// 更新offsetif (isIncrementalMode) {nextSourceOffset = resultSet.getString(offsetColumn);} else {nextSourceOffset = initialOffset;}} else {// 当超过maxBatchSize时，继续查找最后一秒的数据。if(!nextSourceOffset.equals(initialOffset) && nextSourceOffset.equals(resultSet.getString(offsetColumn))){if(null != record) batchMaker.addRecord(record);}// 后续收尾工作}}return nextSourceOffset;}

第二种场景出现的原因是在同一秒内同时出现写入和查询操作，查询时无法取出应取出的数据。
解决的思路为错峰，通过配置手段将查询动作和写入动作错开。

// oracle
select * from user where update_time < TO_TIMESTAMP('${offset}','yyyy-MM-dd HH24:mi:ss.ff') and update_time < SYSDATE - INTERVAL '1' SECOND order by update_time;
// mysql
select * from user where update_time < '${offset}' and update_time < DATE_SUB(now(), INTERVAL 1 SECOND) order by update_time;
// dm
select * from user where update_time < TO_TIMESTAMP('${offset}','yyyy-MM-dd HH24:mi:ss.ff') and update_time < CURRENT_TIMESTAMP- INTERVAL '1' SECOND order by update_time;
// kingbase
select * from user where update_time < '${offset}' and update_time < current_timestamp - INTERVAL '1' SECOND order by update_time;

需要特别注意：因为数据库中存储的时间有可能为业务服务的时间，要保证数据库和业务服务的时区和时间要保持一致。

通道示意图：

新版Streamsets的布局，我的不长这样。 streamsets-jdbc示意图
源：无特殊配置
Jython处理器：根据源传过来的数据查询目标表，对数据进行标记。
流选择器：根据数据的标记分发数据，标记为insert的走新增通道，标记为update的走修改通道。
目标：一个配置为INSERT，另一个配置为UPDATE。
Jython脚本：

import java.sql.DriverManager as DriverManager
import java.lang.Class as Class
import timeurl = "jdbc:mysql://localhost:3306/db?autoReconnect=true&useSSL=false&characterEncoding=utf8"
Class.forName("com.mysql.jdbc.Driver")
username = "root"
password = "passwd"
batch_size = 1000primary_key = "id"
table_name = "t_target"
ids = []
db_ids = set()
records = sdc.records
conn = None
stmt = None
rs = None
if len(records) != 0:try:conn = DriverManager.getConnection(url,username,password)if conn is not None:stmt = conn.createStatement()start_time = time.time()for record in records:id = record.value[primary_key]ids.append(id)num_batches = len(ids) // batch_size + (1 if len(ids) % batch_size != 0 else 0)for i in range(num_batches):start_index = i * batch_sizeend_index = min((i+1) * batch_size,len(ids))batch_ids = ids[start_index,end_index]sql = "select ' + primary_key + ' from " + table_name + " where '+ primary_key +' in ('"for j,id in enumerate(batch_ids):if j != 0:sql += "','"sql += str(id)sql += "')"rs = stmt.executeQuery(sql)while rs.next():id = rs.getString(primary_key)db_ids.add(id)end_time = time.time()sdc.log.info('from '+ table_name + 'query:' + str(len(ids)) + 'rows cost:'+str(end_time - start_time) + 's')for record in records:id = record.value[primary_key]if id in db_ids:record.value['insert_or_update'] = 'update'else:record.value['insert_or_update'] = 'insert'sdc.output.write(record)except Exception as e:raise RuntimeError(e)finally:if rs:rs.close()if stmt:stmt.close()if conn:conn.close()
else:sdc.log.trace('no more data')