Cassandra作为一款分布式NoSQL数据库，凭借高可用、线性扩展和跨数据中心复制能力，被广泛应用于日志存储、时序数据与用户画像等场景。Python开发者通常使用DataStax官方维护的cassandra-driver来连接和操作Cassandra。这个驱动不仅封装了CQL二进制协议，还提供了连接池、自动节点发现、重试策略、负载均衡以及同步与异步两套执行接口。本文围绕实际业务中的常见需求，深入讲解连接参数调优、预处理语句、分页批量操作以及异步并发写入，帮助你在Python侧把Cassandra性能发挥到位。
一、连接参数与集群发现机制
cassandra-driver的核心入口是Cluster对象，它负责维护与多个Cassandra节点的连接池，并通过Session执行CQL语句。创建Cluster时至少需要指定一个初始联系点contact_points，驱动会通过该节点获取集群元数据，自动发现其余节点。生产中建议同时提供多个种子节点，避免单点故障导致初始化失败。端口默认9042，如果集群开启了客户端到节点加密，需要额外配置SSL选项。
连接参数中另一个容易被忽略的是负载均衡策略。默认的DCAwareRoundRobinPolicy会优先访问本地数据中心节点，减少跨机房延迟。可以通过Cluster的load_balancing_policy参数指定。此外，protocol_version建议显式设置为与Cassandra版本匹配的协议版本，例如Cassandra 3.x使用v4，4.x使用v5。如果不指定，驱动会自动协商，但在网络波动时自动协商过程可能增加握手耗时。
from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from cassandra.policies import DCAwareRoundRobinPolicy
auth_provider = PlainTextAuthProvider(username='cassandra', password='cassandra')
cluster = Cluster(
contact_points=['10.0.0.1', '10.0.0.2'],
port=9042,
auth_provider=auth_provider,
load_balancing_policy=DCAwareRoundRobinPolicy(local_dc='dc1'),
protocol_version=4
)
session = cluster.connect()
rows = session.execute('SELECT release_version FROM system.local')
for row in rows:
print(row.release_version)
cluster.shutdown()
连接建立后，Session内部会维护一个连接池，每个节点默认有1到2个连接，并发高时可以适当增加core_connections_per_host和max_connections_per_host。注意不要盲目调大，因为每个连接都会占用服务端线程和内存资源。对于写入密集场景，增加连接数通常比增加客户端进程更有效。
二、预处理语句与参数绑定
预处理语句是cassandra-driver中提升查询效率的关键机制。使用session.prepare将CQL语句发送到服务端编译，之后每次执行只需发送参数值，省去重复解析和编译开销。对于需要频繁执行的插入或查询，预处理后的语句性能提升非常明显。驱动内部会按查询字符串缓存预处理结果，因此不必担心重复调用prepare产生额外网络请求。
参数绑定使用百分号占位符%s，注意Cassandra的CQL不支持问号占位符。绑定参数时可以使用prepared.bind()生成BoundStatement，也可以在session.execute中直接传入参数元组。后者更简洁，但显式绑定可以复用同一个预处理对象并设置一致性级别等属性。参数值会按照CQL类型自动序列化，例如Python的uuid.UUID对应Cassandra的uuid类型，datetime对应timestamp。
from cassandra.query import SimpleStatement
from cassandra import ConsistencyLevel
query = "INSERT INTO users (user_id, name, age) VALUES (%s, %s, %s)"
prepared = session.prepare(query)
for user in user_list:
bound = prepared.bind((user.id, user.name, user.age))
bound.consistency_level = ConsistencyLevel.LOCAL_QUORUM
session.execute(bound)
预处理语句不仅能防止CQL注入，还能让驱动在服务端维护查询计划，降低每次执行的CPU开销。需要注意的是，预处理语句与Schema绑定，当表结构发生变更后需要重新prepare，否则可能返回InvalidQueryException。可以在捕获该异常后清理缓存并再次准备。
三、分页查询与批量写入优化
Cassandra默认查询结果会一次性返回所有行，但在数据量大时这会导致客户端内存暴涨。通过设置fetch_size可以控制单次获取的行数，驱动会在后台自动拉取下一页。对于需要手动控制分页的场景，可以使用paging_state保存当前页游标，在下一次查询时传入，从而实现跨请求的分页。
批量写入时建议区分原子性需求。如果只是常规的日志写入或数据导入，使用execute_concurrent并发执行单条语句比BatchStatement更高效，因为BatchStatement要求协调节点暂存全部语句并保证原子性，会增加网络和内存压力。只有需要保证同一分区内多行原子写入时才适合使用BatchStatement。
from cassandra.query import SimpleStatement
from cassandra.query import BatchStatement
statement = SimpleStatement("SELECT * FROM events", fetch_size=100)
result = session.execute(statement)
for row in result:
process(row)
if result.has_more_pages:
page_state = result.paging_state
next_result = session.execute(statement, paging_state=page_state)
batch = BatchStatement()
batch.add(prepared.bind(('u1', 'Alice', 30)))
batch.add(prepared.bind(('u2', 'Bob', 25)))
session.execute(batch)
分页查询时要注意，paging_state只能在同一个查询和一致性级别下使用，如果更换节点或改变查询条件，游标会失效。另外，跨页查询期间如果数据发生变化，Cassandra不保证快照隔离，可能读到不一致的数据。对于批量导入，建议将同一个分区的数据合并为一个BatchStatement，不同分区则拆分为多个批次并发执行。
四、异步API与并发性能调优
cassandra-driver提供完整的异步执行能力。同步session.execute会阻塞当前线程直到响应返回，在批量任务中容易成为瓶颈。异步API通过session.execute_async返回ResponseFuture对象，可以配合回调或concurrent模块实现高并发。最方便的是execute_concurrent函数，它接收一个语句列表，内部线程池并发执行，并通过concurrency参数控制同时运行的请求数。
并发数并不是越大越好。Cassandra节点能同时处理的请求数受限于CPU核数、内存和磁盘IO，过高并发会导致节点排队，客户端超时概率反而上升。建议从50到100开始压测，观察P99延迟和服务端pending任务数，逐步调整。实际测试中，将同步循环写入改为execute_concurrent后，写入吞吐量通常可以提升3到5倍。
from cassandra.concurrent import execute_concurrent
statements = []
for i in range(1000):
bound = prepared.bind((f'user_{i}', f'name_{i}', i))
statements.append(bound)
results = execute_concurrent(session, statements, concurrency=100)
success_count = 0
for success, result in results:
if success:
success_count = success_count + 1
else:
print('failed:', result)
print('success:', success_count)
异步执行时要特别注意Session生命周期。调用cluster.shutdown之前必须确保所有ResponseFuture已经完成，否则会抛出NoHostAvailable或连接关闭异常。可以使用wait_for_futures或直接遍历结果列表来等待全部完成。对于长时间运行的服务，建议将Session作为全局单例复用，避免频繁创建连接。
五、常见异常与重试策略
操作Cassandra时最常见的异常是NoHostAvailable，它表示驱动无法从任何节点获得可用连接或所有尝试都失败。该异常的errors属性中记录了每个节点的具体错误原因，例如连接超时、认证失败或协议不匹配。排查时应先检查网络连通性和端口放行，再确认认证信息是否正确。另一个高频异常是OperationTimedOut，通常由服务端压力过大或客户端超时时间过短导致。
驱动内置了默认重试策略，但生产环境建议根据业务容忍度自定义。读超时的重试相对安全，因为读操作幂等；写超时则需要谨慎，盲目重试可能导致数据重复。通过继承RetryPolicy并重写对应方法，可以针对不同操作类型设置最大重试次数和返回策略。RETRY表示在同一节点重试，RETHROW则抛出异常，IGNORE会忽略错误但通常不推荐。
from cassandra.policies import RetryPolicy
from cassandra.cluster import Cluster
class CustomRetryPolicy(RetryPolicy):
def on_read_timeout(self, query, consistency, required_responses,
received_responses, data_retrieved, retry_num):
if retry_num >= 2:
return self.RETHROW
return self.RETRY
def on_write_timeout(self, query, consistency, write_type,
required_responses, received_responses, retry_num):
return self.RETHROW
cluster = Cluster(
contact_points=['10.0.0.1'],
default_retry_policy=CustomRetryPolicy()
)
除了重试，连接断开后的重连也是稳定性的一部分。驱动默认使用指数退避重连，可以通过reconnection_policy调整。对于跨机房网络抖动频繁的场景，建议适当增大客户端超时时间并设置较长的重连间隔，避免瞬时故障导致大量请求失败。
掌握cassandra-driver的连接调优、预处理语句、分页批量以及异步并发，能够显著降低Python应用访问Cassandra的延迟并提高吞吐量。实际项目中还需要结合集群规模、网络拓扑和业务读写模式，持续调整参数并监控客户端与服务端指标，才能获得最佳表现。
Cassandracassandra-driverPython修改时间：2026-08-25 02:05:50