Spark连接Cassandra执行SQL查询无限挂起问题咨询

尝试通过Spark从Cassandra数据库获取数据时，SQL请求始终无法完成，任务提交后持续挂起，日志显示Cassandra连接断开，但直接在cqlsh中执行相同SQL能立即得到结果。

Spark配置代码 #

object ExternalConf {
    var cassandraHost : String = "cassandra_cassandra-001,cassandra_cassandra-002,cassandra_cassandra-003,cassandra_cassandra-004"
    var masterSpark: String ="local[*]"
}

object Spark {
  val session : SparkSession = SparkSession
              .builder()
              .appName("KStreaming")
              // .config("spark.cassandra.connection.host", ExternalConf.cassandraHost) //default value or args
              .config("spark.cassandra.connection.host", "cassandra_node") //preprod
              .config("spark.cassandra.auth.username", "cassandra")
              .config("spark.cassandra.auth.password", "cassandra")
              .config("output.batch.grouping.buffer.size", "50")
              .config("output.batch.size.bytes", "102400")
              .config("spark.driver.maxResultSize", "4g")
              .config("spark.sql.broadcastTimeout",  "1800")
              .master(ExternalConf.masterSpark)
              .getOrCreate();

    session.sql("CREATE OR REPLACE TEMPORARY VIEW dbv2_product_categories USING org.apache.spark.sql.cassandra OPTIONS (table 'dbv2_product_categories', keyspace 'preprod', pushdown 'true')")
    import session.implicits._
}

查询执行代码 #

def extractData(data: RDD[ConsumerRecord[String, String]]) = {
    import Spark.session.implicits._
    data
        .foreach(message => {
            var persistedProductCategory: DataFrame = Spark.session.sql("SELECT * FROM dbv2_product_categories WHERE account_id = '" + accountId + "' AND name = '" + shopifyProduct.product_type + "'")
        })
}

异常情况与测试结果 #

• 日志显示任务提交后持续挂起，出现Cassandra连接断开信息
• 直接在cqlsh执行相同SQL语句立即返回结果：

cqlsh:kiliba> SELECT * FROM dbv2_product_categories WHERE account_id = 'shopifytest_62c44e48be54d0002900bd62' AND name = '';

 account_id | name | id | breadcrumb | parent_id
------------+------+----+------------+-----------

(0 rows)

1. RDD的foreach中执行Spark SQL是错误操作 #

Spark的foreach是在Executor节点上执行的分布式操作，而SparkSession属于Driver节点对象，无法在Executor中直接调用执行SQL。这种跨节点的非法操作会引发网络阻塞、连接异常，最终导致任务挂起。

2. 字符串拼接SQL破坏谓词下推且有风险 #

用字符串拼接生成SQL，不仅存在SQL注入风险，还会导致Cassandra的谓词下推（pushdown）失效——Spark无法将过滤条件下推到Cassandra，可能会拉取全表数据后再做过滤，数据量较大时直接造成任务挂起。

3. Cassandra连接配置可能存在节点可达性问题 #

配置的cassandra_node域名如果仅能被Driver节点解析，Executor节点无法连接Cassandra，就会出现连接断开、任务挂起的情况。

解决步骤 #

（1）重构查询逻辑，避免在foreach中执行SQL

将RDD转换为DataFrame，通过关联Cassandra临时视图实现批量查询，利用Spark的分布式处理能力：

def extractData(data: RDD[ConsumerRecord[String, String]]) = {
    import Spark.session.implicits._
    // 解析RDD提取查询所需字段，转为DataFrame
    val messageDF = data.map(record => {
        // 从record中解析出accountId和productType
        (accountId, shopifyProduct.product_type)
    }).toDF("account_id", "product_type")

    // 关联Cassandra视图，自动触发谓词下推
    val resultDF = messageDF.join(
        Spark.session.table("dbv2_product_categories"),
        Seq("account_id")
    ).where($"name" === $"product_type")

    // 批量处理查询结果
    resultDF.foreach(row => {
        // 处理单条结果逻辑
    })
}

（2）使用参数化查询替代字符串拼接

如果必须逐条查询（不推荐），用参数化方式保证谓词下推生效：

// 用占位符传递参数，避免字符串拼接
val queryDF = Spark.session.sql(
    "SELECT * FROM dbv2_product_categories WHERE account_id = ? AND name = ?"
).bind(accountId, shopifyProduct.product_type)

（3）修复Cassandra连接配置

• 确保cassandra_node域名能被所有Spark节点（包括Executor）解析，或者直接使用Cassandra节点的IP列表
• 检查Cassandra监听地址配置，允许Spark节点访问
• 添加连接超时配置避免无限等待：

.config("spark.cassandra.connection.timeout_ms", "30000")
.config("spark.cassandra.read.timeout_ms", "60000")

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