一文教会你用Apache SeaTunnel Zeta离线把数据从MySQL同步到StarRocks

在上一篇文章中,我们介绍了如何下载安装部署SeaTunnel Zeta服务(3分钟部署SeaTunnel Zeta单节点Standalone模式环境),接下来我们介绍一下SeaTunnel支持的第一个同步场景:离线批量同步。顾名思意,离线批量同步需要用户定义好SeaTunnel JobConfig,选择批处理模式,作业启动后开始同步数据,当数据同步完成后作业完成退出。

下面以MySQL离线同步到StarRocks为例,介绍如何使用SeaTunnel进行离线同步作业的定义和运行。

1. 定义作业配置文件

SeaTunnel使用配置文件来定义作业,在这个示例中,作业的配置文件如下,文件保存路径~/seatunnel/apache-seatunnel-incubating-2.3.1/config/mysql_to_sr.config

#定义一些作业的运行参数,具体可以参考 https://seatunnel.apache.org/docs/2.3.1/concept/JobEnvConfig
env {
	job.mode="BATCH"  #作业的运行模式,BATCH=离线批同步,STREAMING=实时同步
	job.name="SeaTunnel_Job"
	checkpoint.interval=10000 #每10000ms进行一次checkpoint,后面会详细介绍checkpoint对JDBC Source和StarRocks Sink这两个连接器的影响
}
source {
	Jdbc {
    	parallelism=5 # 并行度,这里是启动5个Source Task来并行的读取数据
    	partition_column="id" # 使用id字段来进行split的拆分,目前只支持数字类型的主键列,而且该列的值最好是离线的,自增id最佳
    	partition_num="20" # 拆分成20个split,这20个split会被分配给5个Source Task来处理
    	result_table_name="Table9210050164000"
    	query="SELECT `id`, `f_binary`, `f_blob`, `f_long_varbinary`, `f_longblob`, `f_tinyblob`, `f_varbinary`, `f_smallint`, `f_smallint_unsigned`, `f_mediumint`, `f_mediumint_unsigned`, `f_int`, `f_int_unsigned`, `f_integer`, `f_integer_unsigned`, `f_bigint`, `f_bigint_unsigned`, `f_numeric`, `f_decimal`, `f_float`, `f_double`, `f_double_precision`, `f_longtext`, `f_mediumtext`, `f_text`, `f_tinytext`, `f_varchar`, `f_date`, `f_datetime`, `f_timestamp` FROM `sr_test`.`test1`"
    	password="root@123"
    	driver="com.mysql.cj.jdbc.Driver"
    	user=root
    	url="jdbc:mysql://st01:3306/sr_test?enabledTLSProtocols=TLSv1.2&rewriteBatchedStatements=true"
	}
}
transform {
# 在本次示例中我们不需要做任务的Transform操作,所以这里为空,也可以将transform整个元素删除
}
sink {
	StarRocks {
    	batch_max_rows=10240 # 
    	source_table_name="Table9210050164000"
    	table="test2"
    	database="sr_test"
    	base-url="jdbc:mysql://datasource01:9030"
    	password="root"
    	username="root"
    	nodeUrls=[
        	"datasource01:8030" #写入数据是通过StarRocks的Http接口
    	]
	}
}

2. 作业配置说明

在这个作业定义文件中,我们通过env定义了作业的运行模式是BATCH离线批处理模式,同时定义了作业的名称是"SeaTunnel_Job"。checkpoint.interval参数用来定义该作业过程中多久进行一次checkpoint,那什么是checkpoint,以及checkpoint在Apache SeaTunnel中的作用是什么呢?

2.1 checkpoint

查看官方文档中对Apache SeaTunnel Zeta引擎checkpoint的介绍: https://seatunnel.apache.org/docs/2.3.1/seatunnel-engine/checkpoint-storage#introduction 发现checkpoint是用来使运行在Apache SeaTunnel Zeta中的作业能定期的将自己的状态以快照的形式保存下来,当任务意外失败时,可以从最近一次保存的快照中恢复作业,以实现任务的失败恢复,断点续传等功能。其实checkpoint的核心是分布式快照算法:Chandy-Lamport 算法,是广泛应用在分布式系统,更多是分布式计算系统中的一种容错处理理论基础。这里不详细介绍Chandy-Lamport 算法,接下来我们重点说明在本示例中checkpoint对这个同步任务的影响。

Apache SeaTunnel Zeta引擎在作业启动时会启动一个叫CheckpointManager的线程,用来管理这个作业的checkpoint。SeaTunnel Connector API提供了一套checkpoint的API,用于在引擎触发checkpoint时通知具体的Connector进行相应的处理。SeaTunnel的Source和Sink连接器都是基于SeaTunnel Connector API开发的,只是不同的连接器对checkpoint API的实现细节不同,所以能实现的功能也不同。

2.1.1 checkpoint对JDBC Source的影响

在本示例中我们通过JDBC Source连接器的官方文档https://seatunnel.apache.org/docs/2.3.1/connector-v2/source/Jdbc 可以发现如下内容:

这说明JDBC Source连接器实现了checkpoint相关的接口,通过源码我们可以得知,当checkpoint发生时,JDBC Source会将自己还未处理的split做为状态的快照发送给CheckpointManager进行持久化保存。这样当作业失败并恢复时,JDBC Source会从最近一次保存的快照中读取哪些split还未处理,然后接着处理这些split。

在该作业中通过partition_num=20,会将query参数中指定的sql语句的结果分成20个split进行处理,每个split会生成读取它负责的数据的sql,这个sql是由query中指定的sql再加上一些where过滤条件组成的。这20个split会被分配给5个Source Task进行处理,理想情况下,每个Source Task会分配到4个split。假设在一次checkpoint时每个Source Task都只剩下一个split没有处理,这个split的信息会被保存下来,如果这之后作业挂掉了,作业会自动进行恢复,恢复时每个Source Task都会获取到那个还未处理的split,并接着进行处理。如果作业不再报错,这些split都处理完成后,作业运行完成。如果作业还是报错(比如目标端StarRocks挂了,无法写入数据),最终作业会以失败状态结束。

断点续传:

如果在作业失败后,我们修复了问题,并且希望该作业接着之前的进度运行,只处理那些之前没有被处理过的split,可以使用 sh seatunnel.sh -r jobId来让作业ID为jobId的作业从断点中恢复。

回到主题,checkpoint.interval=10000对于从Mysql中读取数据意味着每过10s,SeaTunnel Zeta引擎就会触发一次checkpoint操作,然后JDBC Source Task会被要求将自己还未处理的split信息保存下来,这里需求注意的是,JDBC Source Task读取数据是以split为单位的,如果checkpoint触发时一个split中的数据正在被读取还未完全发送给下游的StarRocks,它会等到这个split的数据处理完成之后才会响应这次checkpoint操作。这里一定要注意,如果MySQL中的数据量比较大,一个split的数据需要很长的时候才能处理完成,可能会导致checkpoint超时。关于checkpoint的超时时长可以参数https://seatunnel.apache.org/docs/2.3.1/seatunnel-engine/checkpoint-storage, 默认是1分钟。

2.1.2 checkpoint对StarRocks Sink的影响

在Sink连接器的文档上,我们也能看到如下图中的标识:

这个标识代表该Sink连接器是否实现了精确处理一次的语义,如果该标识被选中,说明这个Sink连接器能保证发给它的数据它只会往目标端写入一次,不会漏掉导致目标端数据丢失 ,也不会重复往目标端写入。这一功能常见的实现方式是两阶段提交,支持事务的连接器一般会先开启事务进行数据的写入。当checkpoint发生时,将事务ID返回给CheckManager进行持久化,当作业中的所有Task都响应了CheckManager的checkpoint请求后,第一阶段完成。然后Apache SeaTunnel Zeta引擎会调用AggregateCommit的方法让Sink对其事务进行提交,这个过程被称为第二阶段,第二阶段完成后该次checkpoint完成。如果第二阶段提交失败,作业会失败,然后自动恢复,恢复后会再次从第二阶段开始,要求对事务进行提交,直到该事务提交完成,如果事务一直失败,作业也将失败。

并不是只有实现了exactly-once特性的Sink连接器才能保证目标端的数据不丢失不重复,如果目标端的数据库支持以主键去重,那只要Sink连接器保证发送给它的数据至少往目标端写入一次,无论重复写入多少次,最终都不会导致目标端数据丢失或重复。在该示例中StarRocks Sink连接器即是使用了这种方式,StarRocks Sink连接器会将收到的数据先缓存在内存中,当缓存的行数达到batch_max_rows设置的10240行,就会发起一次写入请求,将数据写入到StarRocks中。如果MySQL中的数据量很小,达不到10240行,那就会在checkpoint触发时进行StarRocks的写入。

3. 运行作业

我们使用Apache SeaTunnel Zeta引擎来运行该作业

cd ~/seatunnel/apache-seatunnel-incubating-2.3.1
sh bin/seatunnel.sh --config config/mysql_to_sr.config

作业运行完成后可以看到如下信息,说明作业状态为FINISHED,读取20w行数据,写入StarRocks也是20w行数据,用时6s。

热门相关:骑士归来   惊世毒妃:轻狂大小姐   修仙界最后的单纯   豪门重生盛世闲女   夫人,你马甲又掉了!