如何解决Google Colab与Cassandra的连接及批量数据插入问题

一、在Colab中部署本地Cassandra服务 #

Cassandra依赖Java环境，先完成环境配置与服务启动：

• 安装Java与Cassandra

# 安装OpenJDK 8
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
# 下载并解压Cassandra 4.1.3版本
!wget https://dlcdn.apache.org/cassandra/4.1.3/apache-cassandra-4.1.3-bin.tar.gz
!tar -xzf apache-cassandra-4.1.3-bin.tar.gz
# 设置环境变量
import os
os.environ['JAVA_HOME'] = '/usr/lib/jvm/java-8-openjdk-amd64'
os.environ['CASSANDRA_HOME'] = '/content/apache-cassandra-4.1.3'
os.environ['PATH'] = os.environ['CASSANDRA_HOME'] + '/bin:' + os.environ['PATH']

• 启动并验证服务

# 后台启动Cassandra
!nohup cassandra -R > cassandra.log 2>&1 &
# 等待服务初始化（约30秒）
import time
time.sleep(30)
# 检查节点状态，出现UN标识即为正常
!nodetool status

二、安装Python Cassandra驱动 #

使用Datastax官方维护的Python驱动连接Cassandra：

!pip install cassandra-driver

三、连接Cassandra并创建Keyspace #

from cassandra.cluster import Cluster

# 连接本地Cassandra集群（默认端口9042）
cluster = Cluster(['127.0.0.1'], port=9042)
session = cluster.connect()

# 创建Keyspace（单节点环境用SimpleStrategy复制策略）
session.execute("""
    CREATE KEYSPACE IF NOT EXISTS my_keyspace
    WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}
""")

# 切换到目标Keyspace
session.set_keyspace('my_keyspace')

四、创建数据表 #

以用户信息表为例，定义主键与字段：

session.execute("""
    CREATE TABLE IF NOT EXISTS user_data (
        user_id INT PRIMARY KEY,
        username TEXT,
        email TEXT,
        signup_date DATE
    )
""")

五、批量插入15000条数据 #

采用分批次批量插入，避免单次批量过大导致性能问题：

from cassandra.query import BatchStatement
from datetime import date
import random
import string

batch_size = 1000  # 每批次插入1000条，分15次完成
total_records = 15000

for i in range(0, total_records, batch_size):
    batch = BatchStatement()
    for j in range(i, min(i + batch_size, total_records)):
        # 生成随机测试数据
        username = ''.join(random.choices(string.ascii_lowercase, k=8))
        email = f"{username}@example.com"
        signup_date = date(2023, random.randint(1, 12), random.randint(1, 28))
        # 添加到批量语句
        batch.add("INSERT INTO user_data (user_id, username, email, signup_date) VALUES (%s, %s, %s, %s)",
                  (j, username, email, signup_date))
    # 执行批量插入
    session.execute(batch)
    print(f"已插入 {min(i + batch_size, total_records)} 条数据")

print("15000条数据插入完成")

六、执行查询操作 #

示例常用查询场景：

# 统计数据表总记录数
count_result = session.execute("SELECT COUNT(*) FROM user_data")
print(f"数据表总记录数：{count_result.one()[0]}")

# 查询指定ID的用户信息
user_result = session.execute("SELECT * FROM user_data WHERE user_id = %s", (100,))
user = user_result.one()
if user:
    print(f"用户ID: {user.user_id}, 用户名: {user.username}, 邮箱: {user.email}, 注册日期: {user.signup_date}")

# 查询前10条用户数据（大数据量下避免全表扫描）
top_users = session.execute("SELECT user_id, username FROM user_data LIMIT 10")
print("\n前10条用户数据：")
for user in top_users:
    print(f"ID: {user.user_id}, 用户名: {user.username}")

七、关闭连接 #

操作完成后关闭会话与集群连接：

session.shutdown()
cluster.shutdown()

常见问题说明 #

• Killercoda Playground插入失败原因：Playground存在资源配额与连接限制，不适合大批量数据插入，全程在Colab部署本地Cassandra可规避该问题
• 连接失败排查：查看Cassandra启动日志!cat cassandra.log，确认服务是否正常启动；检查端口9042是否被占用
• 批量插入优化：Cassandra批量操作优先针对同一分区键的数据，跨分区批量建议缩小批次大小，避免性能损耗

内容的提问来源于stack exchange，提问作者Gustavo Labrador