Spark Structured Streaming多查询数据传递与结果合并实现问询

针对你的需求，无需依赖中间Topic或Sink，Spark Structured Streaming原生支持流式查询的结果复用、查询间依赖，以及多结果合并输出。以下结合你的代码给出具体实现方案：

一、将第一个查询的结果作为第二个查询的输入 #

要实现查询间的数据流传递，核心是将上游查询的结果注册为流式临时视图，下游查询直接读取该视图的流数据即可。这里以上游计算商户总消费，下游基于该结果做二次处理为例：

修改后的代码片段 #

// 1. 上游查询：计算商户总消费，注册为流式视图
StreamingQuery merchantSpendQuery = cardTransactionDTODataset.groupBy("merchant")
        .agg(sum("amount").alias("total_amount"))
        .writeStream()
        .queryName("merchant_total_spends") // 注册为流式临时视图
        .outputMode("complete")
        .format("memory") // 存储视图数据（开发测试用，生产可替换为foreachBatch自定义存储）
        .start();

// 2. 下游查询：读取上游视图的流数据，做二次处理（例如筛选高消费商户）
Dataset<Row> highSpendMerchants = sparkSession.readStream()
        .table("merchant_total_spends")
        .filter(col("total_amount").gt(10000)); // 筛选总消费超过10000的商户

// 输出下游处理结果
highSpendMerchants.writeStream()
        .outputMode("complete")
        .format("console")
        .start();

关键说明 #

• 使用queryName将上游查询注册为流式视图，通过sparkSession.readStream().table()读取视图的流数据；
• memory格式适合开发测试，生产环境若需处理大规模数据，建议用foreachBatch将每个批次的结果写入可靠存储（如Spark SQL临时表），再供下游查询读取；
• 仅complete模式的上游查询适合作为输入（输出全量结果），若上游是append/update模式，下游读取的是增量数据，需调整处理逻辑。

二、合并两个查询的结果并输出到Pulsar Topic #

要合并两个聚合查询的结果，需先统一两个结果集的结构（列名、类型），再通过union合并，最后输出到Pulsar。

修改后的代码片段 #

// 1. 商户聚合：统一列名并添加类型标识
Dataset<Row> merchantAgg = cardTransactionDTODataset.groupBy("merchant")
        .agg(sum("amount").alias("total_amount"))
        .withColumnRenamed("merchant", "group_key") // 将分组列统一命名为group_key
        .withColumn("agg_type", lit("merchant")); // 添加标识，区分是商户/分类聚合

// 2. 分类聚合：统一列名并添加类型标识
Dataset<Row> categoryAgg = cardTransactionDTODataset.groupBy("category")
        .agg(sum("amount").alias("total_amount"))
        .withColumnRenamed("category", "group_key")
        .withColumn("agg_type", lit("category"));

// 3. 合并两个结果集（结构完全一致，可直接union）
Dataset<Row> combinedAgg = merchantAgg.union(categoryAgg);

// 4. 输出合并结果到Pulsar Topic
combinedAgg.writeStream()
        .outputMode("complete")
        .format("pulsar")
        .option("service.url", PULSAR_SERVICE_URL)
        .option("topic", "spark/tutorial/combined-agg-results") // 目标Pulsar Topic
        .option("checkpointLocation", "/tmp/spark-checkpoint/combined-agg") // 必须设置checkpoint保障容错
        .start();

关键说明 #

• 合并前必须确保两个DataFrame的结构完全一致：通过withColumnRenamed统一分组列名，lit添加类型标识列；
• 输出到Pulsar时，必须指定checkpointLocation，这是Structured Streaming实现容错的必要配置；
• 需确保Spark-Pulsar连接器版本与你的Spark版本兼容（例如Spark 3.x对应pulsar-spark-connector_2.12:3.4.0）。

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