如何将Spark SQL DataFrames转换为Structured Streaming DataFrames

Spark 结构化流本身没有直接将已存在的批DataFrame强制转换为Streaming DataFrame的API，你可以通过以下两种适配方案实现需求：

方案1：批数据源直接对接结构化流源（推荐） #

如果你的批数据是存储在外部存储（HDFS、S3、本地磁盘等）的，直接使用结构化流的对应源读取即可，通过参数控制每批次加载的数据量，完全匹配你要的「每批次合并到流DataFrame」的需求，示例代码：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.Trigger;

public class BatchToStream {
    public static void main(String[] args) throws Exception {
        SparkSession spark = SparkSession.builder()
                .appName("BatchToStreamDemo")
                .master("local[*]")
                .getOrCreate();

        // 用结构化流读取批文件源，每次触发读1个新文件作为一批
        Dataset<Row> streamingDF = spark.readStream()
                .format("parquet") // 对应你批数据的存储格式，csv、json都支持
                .option("maxFilesPerTrigger", 1) // 每批次最多加载多少个新文件
                .option("path", "你的批数据存储路径")
                .schema(你批数据的Schema) // 结构化流要求提前指定Schema
                .load();

        // 这里可以直接使用所有结构化流的功能，比如窗口计算、持续作业等
        StreamingQuery query = streamingDF.writeStream()
                .format("console")
                .trigger(Trigger.ProcessingTime("1分钟")) // 每1分钟触发一次批次处理
                .start();

        query.awaitTermination();
    }
}

注意：这个方案会自动识别路径下的新增文件，不会重复处理已经加载过的文件，不需要你手动做批次合并，结构化流内部会维护所有已经处理过的数据状态。

方案2：内存队列适配已有批DataFrame #

如果你已经在Spark会话中生成了多个批处理DataFrame，需要手动把每一批追加到流中，可以用内存队列作为结构化流的源，每生成一个批DataFrame就写入队列，流端从队列读取得到Streaming DataFrame：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.execution.streaming.MemoryStream;
import scala.collection.JavaConverters;

public class InMemoryBatchToStream {
    public static void main(String[] args) throws Exception {
        SparkSession spark = SparkSession.builder()
                .appName("InMemoryDemo")
                .master("local[*]")
                .getOrCreate();

        // 初始化内存流，指定数据Schema
        MemoryStream<Row> memoryStream = new MemoryStream<>(1, spark.sqlContext(), scala.Option.empty(), org.apache.spark.sql.Encoders.row(你的批数据Schema));

        // 获取对应的Streaming DataFrame
        Dataset<Row> streamingDF = memoryStream.toDF();

        // 启动流处理作业
        StreamingQuery query = streamingDF.writeStream()
                .format("console")
                .trigger(Trigger.ProcessingTime("30秒"))
                .start();

        // 模拟每次生成批DataFrame后追加到流
        while (true) {
            // 这里替换为你自己的批处理逻辑生成的DataFrame
            Dataset<Row> batchDF = spark.read().parquet("本次批次的文件路径");
            
            // 将批DataFrame的数据写入内存流
            memoryStream.addData(JavaConverters.asScalaIteratorConverter(batchDF.toLocalIterator()).asScala().toSeq());
            
            // 模拟批次间隔
            Thread.sleep(30000);
        }

        query.awaitTermination();
    }
}

注意事项 #

• 两种方案都要求你提前明确数据的Schema，结构化流默认不支持Schema自动推断，生产环境不建议开启自动推断功能
• 如果你需要对全量合并后的流数据做状态计算，直接用结构化流的状态API即可，不需要手动做批次数据合并，框架内部会自动维护状态。

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