如何将Spark DataFrame并行写入Cassandra？

嘿，这个场景我熟得很！10万条记录并行写入Cassandra，其实有几个非常实用的方案，我给你一步步拆解：

如果你有Spark集群可用，这绝对是最优解——Spark天生就是为并行处理数据设计的，写入Cassandra的效率非常高。

步骤拆解： #

1. 初始化SparkSession并配置Cassandra连接器
   要确保你的Spark环境已经安装了Cassandra连接器（一般是spark-cassandra-connector包）。
2. 将Pandas DataFrame转为Spark DataFrame
3. 配置写入参数并并行写入

代码示例： #

from pyspark.sql import SparkSession

# 初始化SparkSession，替换成你的Cassandra连接信息
spark = SparkSession.builder \
    .appName("CassandraParallelWrite") \
    .config("spark.cassandra.connection.host", "你的Cassandra主机地址") \
    .config("spark.cassandra.connection.port", "9042") \
    .config("spark.cassandra.auth.username", "你的用户名") \
    .config("spark.cassandra.auth.password", "你的密码") \
    .config("spark.cassandra.output.batch.size.rows", "1000")  # 调整批处理行数，根据数据大小调整
    .config("spark.cassandra.output.concurrent.writes", "8")  # 并行写入数，根据集群性能调整
    .getOrCreate()

# 把Pandas DataFrame转成Spark DataFrame
spark_df = spark.createDataFrame(你的PandasDataFrame变量名)

# 写入Cassandra表，替换成你的keyspace和table名
spark_df.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="目标表名", keyspace="目标keyspace") \
    .mode("append")  # 可选append/overwrite/ignore，根据需求选择
    .save()

# 关闭SparkSession
spark.stop()

参数调整建议： #

• spark.cassandra.output.batch.size.rows：建议500-2000条，太大容易触发Cassandra的超时，太小会增加请求次数。
• spark.cassandra.output.concurrent.writes：一般设为8-16，不要超过Cassandra集群的负载能力。

如果没有Spark集群，用Python自带的线程池配合官方的cassandra-driver也能实现高效并行写入，适合小集群或单机场景。

步骤拆解： #

1. 初始化Cassandra连接，预编译插入语句
2. 将DataFrame拆分为小批次
3. 用线程池并行写入每个批次

代码示例： #

from cassandra.cluster import Cluster
from cassandra.query import BatchStatement, PreparedStatement
import concurrent.futures
import pandas as pd

# 初始化Cassandra连接，替换成你的信息
cluster = Cluster(['你的Cassandra主机地址'], port=9042)
session = cluster.connect('目标keyspace')

# 预编译插入语句！一定要用占位符，既安全又提升性能
insert_stmt = session.prepare("""
    INSERT INTO 目标表名 (列1, 列2, 列3)  -- 替换成你的表字段
    VALUES (?, ?, ?)
""")

# 拆分DataFrame为小批次，比如每1000条一个批次
batch_size = 1000
batches = [你的PandasDataFrame变量名[i:i+batch_size] for i in range(0, len(你的PandasDataFrame变量名), batch_size)]

# 定义单批次写入函数
def write_batch(batch_df):
    batch = BatchStatement()
    for _, row in batch_df.iterrows():
        # 替换成你的字段顺序
        batch.add(insert_stmt, (row['列1'], row['列2'], row['列3']))
    session.execute(batch)

# 用线程池并行执行，max_workers根据机器性能调整（8-16比较合适）
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
    executor.map(write_batch, batches)

# 关闭连接
cluster.shutdown()

进阶优化：用execute_concurrent_with_args #

官方驱动自带的execute_concurrent_with_args比手动写线程池更高效，它已经做了异步IO优化：

from cassandra.concurrent import execute_concurrent_with_args

# 把DataFrame转为元组列表
data_tuples = [tuple(row) for _, row in 你的PandasDataFrame变量名.iterrows()]

# 并行执行，concurrency是并行度
execute_concurrent_with_args(session, insert_stmt, data_tuples, concurrency=10)

不管用哪种方案，这些细节能帮你避免踩坑：

• 批次大小要合理：不要贪大，Cassandra的Batch不是为超大规模批量设计的，小批次+并行才是正确姿势。
• 预编译语句必用：Prepared Statement能大幅提升写入性能，还能避免SQL注入风险。
• 调整一致性级别：如果业务允许，把一致性级别设为ONE或LOCAL_ONE，比默认的LOCAL_QUORUM写入速度快很多。
• 监控集群状态：写入时关注Cassandra的pending writes、latency等指标，根据实际负载调整并行度和批次大小。

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