Spark Streaming写入Cassandra失败时如何避免数据丢失并实现重试

我有一个从Kafka持续读取数据的Spark readStream函数，对数据进行若干处理后，希望通过Spark writeStream批量写入Cassandra数据库。写入过程中可能会抛出连接超时等各类异常，请问如何确保数据不丢失，并针对该批次数据执行重试？

我的writeStream函数实现如下：

query = df.writeStream \
    .outputMode("append") \
    .option("checkpointLocation", "path") \
    .option("failOnDataLoss", "false")  \
    .option("maxAttempts", "5") \
    .option("retryOnDataLoss", "true") \
    .option("failedWriteFile", "path") \
    .foreachBatch(save) \
    .start()

对应的save方法：

def save(df, batch_id):
    try:
        (df.write
         .format("org.apache.spark.sql.cassandra")
         .options(table=tableName, keyspace=keyspaceName)
         .mode("append")
         .save())
        return None
    except Exception as e:
        raise e

我原本以为当save方法抛出异常时，Spark会重试该批次直到耗尽重试次数，若仍失败则写入指定路径并继续处理下一批次。但我在Spark官方文档或spark-cassandra-connector库中未找到maxAttempts、retryOnDataLoss、failedWriteFile这些参数的相关参考，请问这些参数是否有效？或是有其他替代方案？

一、关于自定义参数的有效性 #

maxAttempts、retryOnDataLoss、failedWriteFile并非Spark Structured Streaming或spark-cassandra-connector官方支持的配置参数，Spark不会识别这些选项，因此你预期的自动重试、失败数据落地逻辑无法通过这些参数实现。

二、确保数据不丢失+批次重试的替代方案 #

1. 在save方法内手动实现重试逻辑 #

针对Cassandra写入的临时异常（如连接超时、节点不可用），可以在save方法中添加重试机制，结合指数退避避免频繁重试导致集群压力，重试耗尽后将失败批次落地到可靠存储（如HDFS/S3），再抛出异常阻止Spark更新checkpoint，确保后续可恢复数据。

示例代码（使用tenacity库实现重试，也可自行编写循环逻辑）：

import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
# 导入Cassandra常见异常类型
from cassandra.cluster import NoHostAvailable, OperationTimedOut

def save(df, batch_id):
    @retry(stop=stop_after_attempt(5),  # 最多重试5次
           wait=wait_exponential(multiplier=1, min=2, max=10),  # 指数退避：2s→4s→8s→10s→10s
           retry=retry_if_exception_type((NoHostAvailable, OperationTimedOut, Exception)))  # 触发重试的异常类型
    def write_to_cassandra():
        (df.write
         .format("org.apache.spark.sql.cassandra")
         .options(table=tableName, keyspace=keyspaceName)
         .mode("append")
         .save())
    
    try:
        write_to_cassandra()
    except Exception as e:
        # 将失败批次写入容错存储，命名包含batch_id方便后续追溯
        df.write.mode("append").parquet(f"/path/to/failed_batches/batch_{batch_id}")
        # 抛出异常，阻止Spark推进checkpoint，避免数据丢失
        raise e

2. 利用Spark Structured Streaming的Checkpoint机制 #

确保checkpointLocation配置的路径是可靠的分布式存储（如HDFS），当批次写入失败并抛出异常时，Spark不会更新checkpoint中的进度记录。后续重启流任务时，Spark会从checkpoint记录的位置重新处理该批次，避免数据丢失。

注意：

• 建议将failOnDataLoss设置为true（默认值），如果设为false，Spark可能会跳过疑似丢失的数据，导致数据丢失；
• 这种方式依赖任务重启，需确保集群有自动重启机制（如YARN的Application Master自动重启）。

3. 配置spark-cassandra-connector内置重试参数 #

spark-cassandra-connector本身提供了连接层面的重试配置，可在初始化SparkSession时设置，针对Cassandra连接、读写超时等底层异常进行重试：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("CassandraStreamWriter") \
    .config("spark.cassandra.connection.localDC", "your_dc_name") \
    .config("spark.cassandra.connection.timeout_ms", 10000)  # 连接超时时间
    .config("spark.cassandra.read.timeout_ms", 20000)  # 读超时时间
    .config("spark.cassandra.connection.maxRetries", 3)  # 连接失败最大重试次数
    .config("spark.cassandra.connection.retryDelay_ms", 1000)  # 重试间隔时间
    .getOrCreate()

4. 失败批次的离线恢复 #

当批次写入失败并落地到容错存储后，可编写离线任务读取这些失败数据，重新写入Cassandra：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("FailedBatchRecovery").getOrCreate()
# 读取所有失败批次
failed_df = spark.read.parquet("/path/to/failed_batches/batch_*")
# 重新写入Cassandra
failed_df.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(table=tableName, keyspace=keyspaceName) \
    .mode("append") \
    .save()

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