Cassandra性能优化--如何提升交叉分区查询性能

新版效果统计数据在Cassandra已经运行两年了，目前随着维度的拓展以及业务的增长，Cassandra存储的数据量累计已接近2T，风和日丽后的第一场暴风雨终于来临

问题

12月10日下午2点，服务突然无响应，业务线接口出现大量超时，经过定位分析，发现是Cassandra查询请求大量阻塞，进一步观察日志发现是Cassandra多分区查询引起的，虽然这个问题是由业务线的一个小bug引起的（一个in查询包含2w个元素），但同时也暴露出了我们底层服务的瓶颈问题

什么是多分区查询

我们假设id是users表中的分区主键，存储了id为1、2、3、4的4条数据，极端情况下，4条数据会被分配到4个分区进行存储（但也可能在1个分区里），我们假设数据不在一个分区内，然后通过下面的语句一次查询出这4个用户的方式就是多分区查询。

SELECT * FROM my_keyspace.users where id in (1,2,3,4)

什么是固定分区查询

假设order表中有uid, day ,productid三个字段，uid为分区键，day为排序键；我们通过下面SQL在订单表查询出用户1在某几日的消费详情，虽然这里也用到了in操作，但因为指定了uid为1，所以查询请求会只命中一个分区。

SELECT * FROM my_keyspace.orders where uid = 1 and day in ('2019-11-11','2019-11-12','2019-11-13')

分析

假设将A,B,C三条数据存储在一个9节点3副本的集群里，当我们使用SELECT * FROM mykeyspace.mytable WHERE id IN (‘A’,’B’,C’)这样一个查询时，Cassandra的处理机制是这样的:

image.png

客户端与服务端建立同步请求，服务端会根据平衡策略在9台节点中选择一个节点作为调制协调器，负责解析SQL并将请求转发到其它节点，然后拉取对应数据到协调器节点，协调器存储了查询关系和每个数据节点返回的数据，正常情况下当协调器节点获取所有数据后会返回到客户端，相应的如果协调器发生故障，整个查询将根据配置的重试略重新开始请求。
一版情况下多分查询是不会有任何问题的，Cassandra都能够很快的将结果进行返回，但随着业务的变动和数据增长，一次需要查询的分区主键元素会变多，相应的Cassandra需要检索的分区数量也会变大，这样会消耗更多的堆空间，并引发频繁GC导致集群可用性下降。

优化

主要优化方式是将上面请求改为固定分区异步并发请求，上面的SQL为改成下面这种方式多次请求服务端

SELECT * FROM mykeyspace.mytable WHERE id = 'A';
SELECT * FROM mykeyspace.mytable WHERE id = 'B';
SELECT * FROM mykeyspace.mytable WHERE id = 'C';

看到反人类的操作着实让我震惊了一把，这么做不应该更慢吗？看似不合理的操作实则内有玄机。
首先，客户端会与服务端建立session会话，每个session会根据服务器情况设置对应数量的连接池，每个连接池会与服务器建立若干连接，每个连接都是异步的（采用netty异步双工技术实现），所以一个连接是可以同时发出多个请求的，在发送下一个请求前不需要等待上一个请求的完成
这种查询还有另一个好处它不存单一的协调器节点了，查询分摊到了多个cassandra节点上，充分利用了集群的CPU和内存资源

image.png

什么是会话
session是连接的管理，提供了丰富的通信API，如 session.executeAsync；也可以通过session监控正在处理的请求以及服务器的状态 session.getState
3.x

        PoolingOptions poolingOptions = new PoolingOptions();
        poolingOptions
                .setConnectionsPerHost(HostDistance.LOCAL, 16, 40) 
                .setMaxRequestsPerConnection(HostDistance.LOCAL, 32768); 
        Cluster.Builder clusterBuilder = Cluster.builder()
                .addContactPoints(
                         ""
                )
                .withPoolingOptions(poolingOptions)
                .withLoadBalancingPolicy(new RoundRobinPolicy()) 
                .withPort(9042);
        Session session = cluster.build().connect();

4.x

Session session = CqlSession.builder().build(); 

#连接方式与平衡策略放在application.conf,并存放在classpath目录下
datastax-java-driver {
  basic.contact-points = ["127.0.0.1:9042"]
  basic {
    load-balancing-policy {
      local-datacenter = datacenter1
    }
  }

什么是连接池
连接池是连接的集合，可指定连接池大小范围，当连接闲置时间过长时会被自动关闭，直到配置的最小值

PoolingOptions poolingOptions = new PoolingOptions();
poolingOptions.setConnectionsPerHost(HostDistance.LOCAL, 16, 40)

什么是连接
与Cassandra实际通信的是连接，每个连接就是一个netty的异步通信，因为是异步处理的所以每个连接可通过executeAsync方法发送多个request请求（最大支持32K个请求，超出将会阻塞），每个request会生成唯一ID,并返回Future对象（java8异步作业句柄）当所有请求发送完毕后，连接通过ID获取服务端返回的数据

客户端提供了丰富的配置及优化策略如失败重试策略等。篇幅有限这里就不一一列举了

一个完整的例子

import com.datastax.driver.core.*;
import com.datastax.driver.core.policies.RoundRobinPolicy;
import java.io.*;
import java.util.ArrayList;
import java.util.List;
public class CassandraTest {

    public static void main(String[] args) throws IOException {

        String ids = "";
        InputStream is = CassandraTest.class.getClass().getResourceAsStream("/ids.txt");
        BufferedReader br = new BufferedReader(new InputStreamReader(is));
        String[] idArray = ids.split(",");
        System.out.println(idArray.length);

        PoolingOptions poolingOptions = new PoolingOptions();
        poolingOptions
                .setConnectionsPerHost(HostDistance.LOCAL, 16, 40) 
                .setMaxRequestsPerConnection(HostDistance.LOCAL, 32768); 
        Cluster.Builder clusterBuilder = Cluster.builder()
                .addContactPoints(
//  有几个写几个            " "
                )
                .withPoolingOptions(poolingOptions)
                .withLoadBalancingPolicy(new RoundRobinPolicy()) 
                .withPort(9042);
  
        Cluster cluster = null;
        try {
            cluster = clusterBuilder.build(); 
            Session session = cluster.connect();  // 整个客户端共享一个session 看起来比较重
           //如果只请求一次 可以使用SimpleStatement
            PreparedStatement statement = session.prepare(
                    "select id,pv,uv,spv,fpv from bigdata.data where id = ? and product='tk' and day in('2019-12-10','2019-12-09','2019-12-08','2019-12-07','2019-12-06','2019-12-05','2019-12-04','2019-12-03')");
            List<ResultSetFuture> futures = new ArrayList<>();
            long start = System.currentTimeMillis(); 
            for (String id : idArray) { 
                ResultSetFuture resultSetFuture = session.executeAsync(statement.bind(id.trim()));
                futures.add(resultSetFuture);
                 
            } 
            List<Row> results = new ArrayList<>();
            for (ResultSetFuture future : futures) {
                ResultSet rows = future.getUninterruptibly();

                results.addAll(rows.all());
            }
            System.out.println(System.currentTimeMillis() - start);
            System.out.println(results.subList(1, 5));
            System.out.println(results.size());
 
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

写在最后

多分区数据查询是比较消耗性能的，类似的还有allow filtering无法查询，如果需要一次查询的分区键元素数量不是很多，固定分区遍历查询与In关键字查询性能区别不是很大，在我们当前业务环境中通过测试发现，当一次查询元素个数超过1000时固定分区查询性能提升2倍，3000-5000时性能提升3-5倍，待查询的元素个数越多性能提升越明显，整体来看有3-10倍的性能提升，但响应时间也会相应变长。