Spark Streaming与SparkSession共存时Dataset.groupBy().count()返回空表问题

首先，你遇到的这个问题核心是Spark上下文冲突导致的——同时创建独立的SparkSession和JavaStreamingContext实例，哪怕开了allowMultipleContexts，也会让数据处理的上下文混乱，这是Spark 2.x里非常典型的坑。

为什么会出现这种诡异情况？ #

你看到ds.show()有数据，但ds.count()返回0、groupBy后是空表，本质是因为：

• 你初始化了两个完全独立的Spark上下文：一个属于SparkSession，另一个属于JavaStreamingContext。虽然配置了允许多上下文共存，但这会让底层的RDD分区、任务调度出现冲突——读出来的Dataset表面上能打印内容，但它的底层RDD并没有正确绑定到当前Streaming的任务上下文，后续的action操作（比如count、groupBy后的show）根本没真正执行数据计算。
• 在foreachRDD里直接用外部的spark实例，没有和当前处理的RDD绑定，导致数据的上下文不匹配，相当于你在一个上下文里读了数据，却在另一个上下文里去计算，自然拿不到结果。

具体修复步骤 #

1. 统一Spark上下文，彻底抛弃多实例 #

不要分别创建SparkSession和JavaStreamingContext的独立配置，应该从SparkSession衍生出StreamingContext，确保整个应用只有一个上下文：

// 先创建唯一的SparkSession
SparkSession spark = SparkSession 
    .builder() 
    .master("local[4]") 
    .appName(AppName) 
    .config("spark.cassandra.connection.host", ip) 
    .config("spark.cassandra.connection.port", port)
    // 删掉allowMultipleContexts，因为现在只有一个上下文了
    .getOrCreate();

// 从SparkSession的SparkContext生成StreamingContext
SparkConf sparkConfig = spark.sparkContext().getConf();
JavaStreamingContext jssc = new JavaStreamingContext(sparkConfig, batchInterval);

2. 在foreachRDD内部使用当前RDD绑定的SparkSession #

在foreachRDD里，绝对不要直接用外部的spark对象，而是通过当前处理的RDD获取对应的Session，确保上下文完全匹配：

logLines.foreachRDD(rdd -> { 
    // 先完成数据写入Cassandra的操作
    javaFunctions(rdd).writerBuilder("my_keyspace", "table_name", mapToRow(Table.class)).saveToCassandra(); 

    // 获取当前RDD关联的SparkSession，这才是正确的上下文
    SparkSession currentSpark = SparkSession.builder()
        .config(rdd.sparkContext().getConf())
        .getOrCreate();

    // 用当前Session读取Cassandra数据
    Dataset<Row> ds = currentSpark.read() 
        .format("org.apache.spark.sql.cassandra") 
        .options(new HashMap<String, String>() { { 
            put("keyspace", "my_keyspace"); 
            put("table", "table_name"); 
        } }).load().where("theColumn == number").cache(); 

    // 可以先触发count确保数据被正确加载
    long rowCount = ds.count();
    System.out.println("Loaded rows: " + rowCount);
    ds.show(); 

    Dataset<Row> ds2 = ds.groupBy("theColumn").count().cache(); 
    ds2.show(); // 现在应该能正常输出统计结果了
});

3. 额外检查过滤条件的类型匹配 #

虽然你说ds.show()有数据，但还是要确认where("theColumn == number")的类型是否正确：

• 如果theColumn是字符串类型，要写成where("theColumn == 'your_number'")（加单引号）；
• 如果是数值类型，确保number是对应的int/long，避免因类型不匹配导致隐性过滤（比如字符串和数字比较会返回false）。

4. 关于cache的小提示 #

有时候cache操作需要触发action才会真正生效，所以在cache后加个ds.count()能确保数据被缓存到内存，避免后续操作重复读取Cassandra。

最后提个建议 #

Spark 2.2.0里的传统DStream API和SQL/Dataset混合使用本来就容易出问题，如果你有机会升级的话，建议直接用Structured Streaming——它原生支持Dataset/SQL操作，完全不需要处理上下文冲突的问题，是Spark官方推荐的流式处理方案。

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