简析Cassandra的BATCH操作

本文涉及的产品
云数据库 MongoDB,独享型 2核8GB
推荐场景:
构建全方位客户视图
云原生多模数据库 Lindorm,多引擎 多规格 0-4节点
云数据库 Tair(兼容Redis),内存型 2GB
简介: cassandra中批量写入的操作称为batch,通过batch操作可以将多个写入请求合并为一个请求。这样有如下作用: 把多次更新操作合并为一次请求,减少客户端和服务端的网络交互。 batch中同一个partition key的操作具有隔离性。

cassandra中批量写入的操作称为batch,通过batch操作可以将多个写入请求合并为一个请求。这样有如下好处:

  • 把多次更新操作合并为一次请求,减少客户端和服务端的网络交互。
  • batch中同一个partition key的操作具有隔离性。
  • 默认的LOGGED类型可以保证batch中的所有操作要么(最终)全部成功,要么全部失败。也就是原子性。

本文介绍batch的使用和注意事项。并分析batch操作原子性的实现。

batch使用

cqlsh中使用batch

在cqlsh中如下语句可以提交一个batch(例子来自cassandra官方文档):

BEGIN BATCH
   INSERT INTO users (userid, password, name) VALUES ('user2', 'ch@ngem3b', 'second user');
   UPDATE users SET password = 'ps22dhds' WHERE userid = 'user3';
   INSERT INTO users (userid, password) VALUES ('user4', 'ch@ngem3c');
   DELETE name FROM users WHERE userid = 'user1';
APPLY BATCH;

在java应用中使用batch

下面的例子示意了如何使用java客户端实现批量写入:

    PreparedStatement statement = session.prepare("insert into test(intcol,boolcol) values(?,?)");
    BatchStatement batchStatement = new BatchStatement(Type.UNLOGGED);
    for( int i = 0; i < 5; i++ ) {
      BoundStatement boundStmt = statement.bind();
      boundStmt.setInt(0, i);
      boundStmt.setBool(1, false);
      batchStatement.add(boundStmt);
    }
    session.execute(batchStatement);

注意事项

  • UNLOGGED/COUNTER batch操作不保证原子性,可能会部分成功。
  • batch大小(指的是所有statement的内容总的大小)不能超过batch_size_fail_threshold_in_kb,否则会失败;另外,如果batch大小如果超过batch_size_warn_threshold_in_kb,batch会成功,但会收到一个警告提示batch过大,如果日志中频繁看到这样的警告,需要考虑batch的使用是否正确。
  • 不建议在UNLOGGED batch中包含多个partition key,因为这样虽然客户端和服务端之间的交互变少,但是coordinator和其他节点的交互并没有减少。如果partition key的数量超出了unlogged_batch_across_partitions_warn_threshold,会收到警告提示。顺便说一句,cassandra的copy命令在写入时用的就是UNLOGGED batch,在调用batch之前对请求按照key所在的节点做分组,然后再对每个分组调用batch。
  • 表的gc_grace_seconds配置会影响batch的回放。在回放batch时,会检查涉及的所有表的gc_grace_seconds,如果当前时间已经超出了其中某个表的gc_grace_seconds,则整个batch都不会回放直接删除。另外,如果表的gc_grace_seconds为0,会收到警告提示。
  • 如果在batch中使用了条件更新,需要注意只有所有更新操作的条件都满足时batch才会成功,只要有一个不满足batch就会失败;另外,所有更新都只能针对同一个partition key,这是因为cassandra中cas操作只能提供partition级别的一致性保证。

关于batch的原子性

UNLOGGED和COUNTER类型的batch是不保证原子性的,这种batch类型的处理是简单的把写请求分发到对应的节点上面,所以可能会出现部分成功的情况。以下分析LOGGED batch的处理,说明batch是如何做到原子性的。
简单来说,LOGGED batch的处理分三个步骤:
第一步 写入batches表
batches表的结构如下:

                "CREATE TABLE %s ("
                + "id timeuuid,"
                + "mutations list<blob>,"
                + "version int,"
                + "PRIMARY KEY ((id)))")

我们可以看出一个batch的所有mutations都在一行记录里面。这样的设计保证了batchlog写入的原子性,要么成功,要么失败。
第二步 分发mutation
这一步处理和UNLOGGED batch差不多,就是把每个mutation操作发送到他对应的节点上面去。比UNLOGGED batch多出的一个步骤是,如果所有mutation都成功执行了,会删除到batches表中对应的记录。
第三部 回放batchlog
如果第二步失败了,cassandra仍然可以保证batch最终成功。有一个线程池会以10s执行一次replayFailedBatches回放batch。如果batch中的请求全部执行成功就会删除batch。
一些细节:

  • batch中的mutate要么成功发到endpoint,要么成功写入hint并落盘然后才会被删除。
  • 为了避免回放正在执行中的batch,回放时只会选择BATCHLOG_REPLAY_TIMEOUT之前的记录。BATCHLOG_REPLAY_TIMEOUT的值可以通过系统属性“cassandra.batchlog.replay_timeout_in_ms”来设置,如果没有设置,默认值write_request_timeout_in_ms的两倍。

从上面的分析可以看出,batch的原子性主要体现在:

  • 如果batch log写入失败,那么batch的操作全部失败。
  • 如果batch log写入成功,即使某个mutation没有成功,cassandra也会(在gc_grace_seconds之前)一直回放batch,保证最终全部成功。

入群邀约

为了营造一个开放的 Cassandra 技术交流,我们建立了微信群公众号和钉钉群,为广大用户提供专业的技术分享及问答,定期开展专家技术直播,欢迎大家加入。另外阿里云提供免费Cassandra试用:https://www.aliyun.com/product/cds

8c677711345d2450b80800d50aab3fa0b5a95ded

钉钉群入群链接:https://c.tb.cn/F3.ZRTY0o

微信群公众号:
2a9d5b6c3b9030d06ddb59131aab7287945ead53

目录
相关文章
|
存储 分布式数据库 Hbase
HBase scan过程简析
HBase scan过程简析。 scan过程总体上是分层处理的,与存储上的组织方式一致,脉络比较清晰; 具体来说,就是region->store→hfile/memstore,分别都有对应的scanner实现进行数据读取; scan请求本身设置的条件,以及server和table层面的一些参数限制,会根据需要分布在不同层次的scanner中进行处理; 2.
2198 0
HBase scan过程简析
|
6月前
|
SQL 存储 调度
从 Volcano 火山模型到 Pipeline 执行模型,阿里云数据库 SelectDB 内核 Apache Doris 执行模型的迭代
一个合适的执行模型对于提高查询效率和系统性能至关重要。本文全面剖析 Apache Doris Pipeline 执行模型的设计与改造历程,并在 2.1 版本对并发执行模式与调度模式进一步优化,解决了执行并发受限、执行及调度开销大等问题。
从 Volcano 火山模型到 Pipeline 执行模型,阿里云数据库 SelectDB 内核 Apache Doris 执行模型的迭代
|
3月前
|
机器学习/深度学习 缓存 PyTorch
异步数据加载技巧:实现 DataLoader 的最佳实践
【8月更文第29天】在深度学习中,数据加载是整个训练流程中的一个关键步骤。为了最大化硬件资源的利用率并提高训练效率,使用高效的数据加载策略变得尤为重要。本文将探讨如何通过异步加载和多线程/多进程技术来优化 DataLoader 的性能。
386 1
|
3月前
|
Java 关系型数据库 MySQL
实时计算 Flink版操作报错合集之在使用批处理模式中使用flat_aggregate函数时报错,该如何解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
3月前
|
存储 异构计算
自研分布式训练框架EPL问题之通过strategy annotation实现流水并行如何解决
自研分布式训练框架EPL问题之通过strategy annotation实现流水并行如何解决
|
NoSQL Redis Anolis
性能优化特性之:Redis批处理pipeline模式
本文介绍了一种更贴近实际使用的redis验测方法:多pipline模式,并从原理、使用方法进行详细阐述。
|
6月前
|
机器学习/深度学习 消息中间件 人工智能
机器学习PAI报错问题之读取kafka数据报错如何解决
人工智能平台PAI是是面向开发者和企业的机器学习/深度学习工程平台,提供包含数据标注、模型构建、模型训练、模型部署、推理优化在内的AI开发全链路服务;本合集将收录PAI常见的报错信息和解决策略,帮助用户迅速定位问题并采取相应措施,确保机器学习项目的顺利推进。
|
6月前
|
监控 Java 流计算
读Flink源码谈设计:Metric
前阵子笔者涉及了些许监控相关的开发工作,在开发过程中也碰到过些许问题,便翻读了Flink相关部分的代码,在读代码的过程中发现了一些好的设计,因此也是写成文章整理上来。
400 0
读Flink源码谈设计:Metric
|
人工智能 NoSQL Java
SpringBoot实战(十七):Redis Pipeline 轻松实现百倍性能提升(续)
SpringBoot实战(十七):Redis Pipeline 轻松实现百倍性能提升(续)
330 0
|
机器学习/深度学习 PyTorch 算法框架/工具
PyTorch并行与分布式(三)DataParallel原理、源码解析、举例实战
PyTorch并行与分布式(三)DataParallel原理、源码解析、举例实战
790 0
下一篇
无影云桌面