在ScyllaDB，我们开发了高性能的NoSQL数据库开云体育电动老虎机“锡拉”， api兼容Apache Cassandra, Amazon DynamoDB和Redis。今年早些时候，我们引入了对变更数据捕获在锡拉4.3中。这个新特性似乎与Apache Kafka生态系统的集成完美匹配，所以我们开发了锡拉疾控中心源连接器使用Debezium开云体育官方注册网址框架。在这篇博文中，我们将介绍Scylla的CDC的基本结构，我们选择Debezium框架的原因以及我们所做的设计决策。开云体育官方注册网址
疾控中心支持"锡拉"
变更数据捕获(CDC)允许用户跟踪Scylla数据库中的数据修改。开云体育电动老虎机它可以很容易地在任何Scylla表上启用/禁用。在打开它时，将创建所有修改(insert、UPDATEs、delete)的日志并自动更新。
当我们在Scylla中设计CDC的实现时，我们希望使CDC日志易于使用。因此，CDC日志被存储为常规的Scylla表，任何现有的CQL驱动程序都可以访问。当对启用了CDC的表进行修改时，有关该操作的信息将保存到CDC日志表中。
下面是它的一个快速演示:首先，我们将创建一个启用CDC的表:
创建表格ks.orders（user文本, order_idint, order_name文本，主要的关键(user, order_id)) WITH cdc = {＇启用＇：真正的};接下来，让我们执行一些操作:
插入成ks.orders（user, order_id, order_name)值（＇蒂姆＇，1，＇苹果＇）;插入成ks.orders（user, order_id, order_name)值（＇爱丽丝＇，2，＇蓝莓＇）;更新ks.orders集order_name =＇菠萝＇在哪里用户=＇蒂姆＇和order_id =1；最后，让我们看看修改后的表格内容:
选择＊从ks.orders;用户| order_id | order_name------+----------+-------------蒂姆|1|菠萝爱丽丝|2|蓝莓仅看这个表，您无法重建发生的所有修改，例如，更新之前Tim的订单名。让我们看看CDC日志，很容易以表格的形式访问(为了清晰起见，有些列被截断了):
选择＊从ks.orders_scylla_cdc_log;Cdc $stream_id | Cdc $time | | Cdc $operation | order_id | order_name |用户---------------+----------+-...-+---------------+----------+-------------+----------0 x2e46a．.．|7604．.．|……|2|1苹果|蒂姆0 x2e46a．.．|8 f直流……|……|1|1|菠萝|蒂姆0 x41400．.．|808e……|……|2|2|蓝莓|爱丽丝这三个操作在CDC日志中都是可见的。有两个insert (Cdc $operation = 2)和一个UPDATE (Cdc $operation = 1)．对于每个操作，它的时间戳也保存在美国疾病控制与预防中心时间美元列。类指定的基于时间的UUID值将时间戳编码为RFC 4122可以使用Scylla驱动程序中的辅助方法进行解码。
选择Debeziu开云体育官方注册网址m
当我们考虑客户如何访问CDC日志时，将其与Kafka一起使用似乎是最容易访问的方法。因此，我们决定为锡拉疾控中心开发一个源连接器。
对于我们的第一个概念验证，我们使用Kafka连接API．这个原型对我们确定连接器是否可以水平扩展(稍后将在本文中描述)至关重要。
然而，我们很快意识到，如果只使用Kafka Connect API，我们将不得不重新实现许多已经存在于其他连接器中的功能。我们也希望我们的连接器成为Kafka社区的好公民，坚持最佳实践和惯例。这正是我们选择Debezium的原因!开云体育官方注册网址
事件表示
在我们决定使用Debezium框架之后，我们研究了Scy开云体育官方注册网址lla CDC操作应该如何用Debezium的Envelope格式表示。
信封格式由以下字段组成:
人事处-操作类型:c为创建、u对于更新,d对于删除和r为读之前-事件发生前的行状态后-事件发生后的行状态源-事件元数据ts_ms-连接器处理事件的时间
将“锡拉”操作映射到人事处场地相当简单:c插入,u对于更新,d删除。
我们决定跳过跨多行的DELETE事件，比如range DELETE:
删除从ks.table在哪里pk =1和ck >0和ck <5表示这样的操作会不必要地使格式复杂化，以容纳额外的范围信息。此外，它还打破了Envelope表示对单行修改的期望。
在Scylla CDC中，范围delete在CDC表中表示为两行:第一行编码关于被删除范围开始的信息(在上面的例子中:Pk = 1, ck > 0)，第二行编码被删除范围的结束(在上面的例子中:Pk = 1, ck < 5)．关于该范围内的每一行的信息不会被持久化。这与Scylla中的delete会在数据库中生成一个墓碑的事实相对应。开云体育电动老虎机
默认情况下，Scylla的CDC存储主键，只存储操作的修改列。例如，假设我创建了一个表并插入了一行:
创建表格ks。示例(pkintv1intv2int, v3int，主要的关键(pk))与cdc = {＇启用＇：真正的};插入成ks。示例(pk, v1, v2, v3)值（1，2，3.，4）;在Scylla中，您可以发出另一个INSERT语句，它将覆盖一些列:
插入成ks。示例(pk, v1, v3)值（1，20.，零）;的v2在这个查询之后，列保持不变，我们没有关于它之前值的任何信息。
我们必须能够表示三种可能性:一个列没有被修改，一个列被赋值零值或列被分配非空值。我们选择的表现形式的灵感来自开云体育官方注册网址卡桑德拉Debezium连接器，它的工作原理是将列的值包装在结构中:
＂v1＂:{＂价值＂：1}，＂v2＂：零，＂v3＂:{＂价值＂：零｝一个零结构值表示未修改列(v2字段)。如果列被赋值为零值(v3字段)，将会有一个带有零价值字段。非空列赋值(v1字段的内容价值字段。这样的格式允许我们正确地表示所有的可能性，并区分分配零和non-modification。
但是，大多数接收器连接器不能正确解析这样的结构。因此，我们决定在Debezium的基础上开发自己的SMT开云体育官方注册网址新记录状态提取．我们的ScyllaExtractNewStateSMT工作通过应用Debezium的新记录状开云体育官方注册网址态提取和扁平化{“价值”:}结构(以不能区分为代价零值和缺失的列值):
＂v1＂：1，＂v2＂：零，＂v3＂：零“锡拉”的疾病预防控制中心还支持在每次操作时记录预图像和后图像(需额外付费)。我们计划在未来版本的Scylla CDC源连接器中添加对它们的支持。
水平扩展
即使在概念验证阶段，出色的性能也是最重要的要求。Scylla数开云体育电动老虎机据库可以扩展到数百个节点和PBs的数据，因此很明显，单个Kafka Connect工作节点(甚至是多线程的)无法处理大型Scylla集群的负载。
值得庆幸的是，我们在Scylla中实现CDC功能时考虑到了这一点。通常，您可以将变更数据捕获看作是按时间顺序排列的变更队列。为了允许水平扩展，Scylla维护了一组多个按时间顺序排列的更改队列，称为流。当CDC日志只有一个消费者时，它必须查询所有流以正确读取所有更改。这种设计的一个好处是，您可以引入额外的消费者，为每个消费者分配一个流的分离集。因此，您可以极大地提高处理CDC日志的并行性。
这就是我们在Scylla CDC源连接器中实现的方法。启动时，连接器首先读取所有可用流的标识符。接下来，它将它们分配到许多Kafka Connect任务中(可通过tasks.max)．
每个创建的Kafka Connect任务(可以在单独的Kafka Connect节点上运行)从其分配的流集中读取CDC更改。如果将任务数量增加一倍，每个任务只需读取流数量的一半——即数据吞吐量的一半，从而可以处理更高的负载。
解决大流量问题
在Scylla中设计CDC功能时，我们必须仔细选择要创建的流的数量。如果我们选择的流太少，消费者可能无法跟上单个流的数据吞吐量。这也会降低INSERT、UPDATE、DELETE操作的速度，因为许多并发操作会争夺对单个流的访问权。然而，如果Scylla创建了太多的流，消费者将不得不向Scylla发出大量的查询(以覆盖每个流)，从而造成不必要的负载。
目前在"锡拉"中实施的疾控中心Number_of_nodes * number_of_vnode_per_node * number_of_shards每个集群的流。vnode的数量是指Scylla使用了环结构，默认每个节点有256个vnode。每个Scylla节点由几个独立的碎片组成，这些碎片包含它们在节点总数据中的份额。通常，每个超线程或物理内核都有一个碎片。
例如，如果您创建了一个4节点的i3。金属(每个节点72个vCPU) Scylla集群，每秒能够进行大约600k次操作(一半insert，一半select)，即:4 * 256 * 72 = 73728流。
我们很快意识到，在更大的集群中，这么多流可能是个问题:
对Scylla的查询太多了——每个流一个查询
太多的Kafka连接偏移量-每个流一个偏移量。存储偏移量意味着连接器可以在崩溃后从上次保存的位置恢复。
为了减轻这些问题，我们决定在客户端对流进行分组。我们选择按VNode对流进行分组。这样就减少了计数Number_of_nodes * number_of_vnode_per_node * number_of_shards来Number_of_nodes * number_of_vnodes_per_node．对于4节点的i3。金属，这意味着从73728减少到1024:只有1024个查询Scylla和1024个偏移存储在Kafka。
然而，我们仍然对Kafka上存储的偏移量感到不安。当我们查看其他连接器时，大多数连接器在每个复制表中只存储一个偏移量，或者最多存储几十个偏移量(因此可伸缩性有限)。
为了理解为什么在Kafka Connect上存储成千上万的数据流会是个问题，让我们看看它是如何工作的。源连接器创建的每个Kafka Connect记录都包含一个键/值偏移量，例如:key -my_table；抵消,25，这可以表示连接器完成了25行的读取my_table．定期(由offset.flush.interval.ms)，这些偏移量被刷新到Kafka主题connect-offsets，作为常规的Kafka消息。
不幸的是，Kafka不是一个键/值存储。连接器启动时，必须扫描上的所有消息connect-offsets来找到它需要的那个。当它更新先前保存的偏移量时，它只是将新值追加到此主题，而不删除先前的条目。对于只有一个偏移量的连接器来说，这不是问题——当每分钟更新一次时，这个主题一周后将保存大约10,000条消息。然而，在锡拉疾控中心源连接器的情况下，这个数字可能会大几个数量级!
幸运的是，通过在上设置更激进的压缩配置，可以轻松缓解此问题connect-offsets的话题。的默认配置retention.ms7天之后segment.bytes从1GB，这个主题可以在几个小时后增长到几百兆字节(Scylla集群有几十个节点，非常小)offset.flush.interval.ms)．这使得连接器启动时间变慢，因为它必须在启动/重新启动后扫描整个偏移主题。通过调优segment.bytes，segment.ms或cleanup.policy，retention.ms我们能够缓解这个问题，并显著降低connect-offsets主题的大小。前两个选项指定日志压缩过程的频率。当一个段被压缩时，所有具有相同键的消息都被缩减为最新的消息(最新的偏移量)。或者，设置较短的保留时间(但要大于Scylla的CDC保留时间)被证明是减小偏移主题大小的好选择。
基准:接近线性缩放
为了验证我们的连接器实际上可以水平扩展，我们执行了一个基准测试，以测量Scylla CDC源连接器在越来越大的Kafka Connect集群上的最大吞吐量。
首先，我们启动了一个单节点i3.4 .4xlarge Scylla集群(基于官方Scylla AMI)。接下来，我们将5000万行(总大小5.33GB)插入到一个启用cdc的表中。后来，我们在1,3或5个节点(r5n.2xlarge)上启动了Apache Kafka 2.6.0集群和Kafka Connect集群。我们启动Scylla CDC源连接器，从先前填充的CDC启用表中消费数据，并测量产生所有5000万条Kafka消息所需的时间。
我们的连接器能够几乎线性地扩展吞吐量:
| Kafka集群大小 | 吞吐量 | 加速 |
|---|---|---|
1个节点 |
46 k / s |
1 x |
3个节点 |
129 k / s |
2.8倍 |
5个节点 |
215 k / s |
4.7倍 |
结论
在这篇博文中，我们深入研究了Scylla CDC源连接器的开发。我们从疾控中心在“锡拉”中的实施概述开始。我们已经讨论了选择Debezium而不是Kafka Connect API来构开云体育官方注册网址建连接器的原因，从而使它为用户所熟悉并成为Kafka的惯用方法。接下来，我们研究了遇到的两个问题:如何表示Scylla更改以及如何使连接器可伸缩。
我们非常高兴能够继续改进连接器，提供更多的功能，使其性能更好。我们热切地期待着Debezium生态系统的发展，并集成在Debezium最新版本中引入的功能。开云体育官方注册网址
如果你想自己检查连接器，GitHub存储库及其源代码可在这里:github.com/scylladb/scylla-cdc-source-connector．你可以在这里了解更多关于Scylla的信息:scylladb.com．
关于Debe开云体育官方注册网址zium
开云体育官方注册网址Debezium是一个开源的分布式平台，它将现有数据库转换为事件流，因此应用程序几乎可以立即看到并响应数据库中提交的每一个行级更改。开云体育电动老虎机开云体育官方注册网址Debezium是建立在卡夫卡并提供卡夫卡连接监控特定数据库管理系统的兼容连接器。开云体育电动老虎机开云体育官方注册网址Debezium在Kafka日志中记录了数据更改的历史，所以你的应用程序可以在任何时候停止和重新启动，并且可以很容易地使用它没有运行时错过的所有事件，确保所有事件都被正确和完整地处理。开云体育官方注册网址Debezium是开源下Apache许可证，版本2.0．
参与
我们希望您觉得Debezium有趣开云体育官方注册网址且有用，并愿意尝试一下。在Twitter上关注我们@开云体育官方注册网址debezium，在Zulip上和我们聊天，或加入我们的邮件列表与社区对话。所有的代码都是开源的GitHub上，因此在本地构建代码并帮助我们改进现有连接器并添加更多连接器。如果您发现了问题或对我们如何改进Debezium有想法，请告诉我们开云体育官方注册网址记录问题．