如何将Seq[String]字符串序列转换为Spark流式DataFrame？

针对生产环境下把内存中的Seq[String]转为流式DataFrame的需求，这里提供两种可行方案，避开仅适用于测试的MemoryStream和生成静态DataFrame的createDataFrame：

方案1：基于文件流（通用生产方案） #

这是Spark流式处理中最稳定的通用方案，通过将Seq数据写入临时目录，再用文件流监听该目录实现流式读取，后续还能追加新数据到流中。

步骤1：写入初始Seq数据到临时目录 #

import org.apache.spark.sql.SparkSession
import java.nio.file.{Files, Paths}

val spark = SparkSession.builder()
  .appName("SeqToStreamingDF")
  .master("local[*]") // 生产环境可移除该配置
  .getOrCreate()
import spark.implicits._

// 待转换的字符串序列
val rawSeq = Seq("spark", "streaming", "production", "ready")

// 创建临时目录（自动生成唯一目录）
val tempDir = Files.createTempDirectory("stream_data").toString

// 将Seq写入临时目录（文本格式）
rawSeq.toDF("content").write.mode("append").text(tempDir)

步骤2：启动文件流读取 #

// 定义流式DataFrame
val streamingDF = spark.readStream
  .format("text")
  .option("path", tempDir)
  .option("maxFilesPerTrigger", 1) // 可选：每次触发处理一个文件，控制流速度
  .load()

// 输出流式结果到控制台（生产环境可替换为Kafka、HDFS等Sink）
val streamQuery = streamingDF.writeStream
  .outputMode("append")
  .format("console")
  .start()

streamQuery.awaitTermination()

追加新数据到流中 #

如果后续需要向流中添加新数据，直接往临时目录写入即可：

val newData = Seq("new", "batch", "data")
newData.toDF("content").write.mode("append").text(tempDir)

方案2：自定义流式数据源（灵活定制） #

如果不想依赖文件系统，可以实现自定义流式数据源，直接从内存Seq中按批次输出数据。这种方式需要处理Offset管理，确保流的语义正确性。

实现自定义数据源 #

import org.apache.spark.sql.execution.streaming.Source
import org.apache.spark.sql.sources.DataSourceRegister
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{DataFrame, SQLContext}

// 自定义Offset，用于跟踪流处理进度
case class SeqOffset(value: Int) extends org.apache.spark.sql.execution.streaming.Offset {
  override def json: String = value.toString
}

// 自定义流式数据源
class SeqStreamSource(sqlContext: SQLContext, params: Map[String, String]) extends Source with DataSourceRegister {
  // 从参数中获取原始Seq数据
  private val dataSeq = params("data").split(",").toSeq
  // 跟踪当前处理到的位置
  private var currentOffset = 0

  // 定义输出Schema
  override def schema: StructType = new StructType().add("content", "string")

  // 获取当前最新的Offset
  override def getOffset: Option[org.apache.spark.sql.execution.streaming.Offset] = {
    if (currentOffset >= dataSeq.length) None else Some(SeqOffset(currentOffset))
  }

  // 获取指定区间的批次数据
  override def getBatch(
    start: Option[org.apache.spark.sql.execution.streaming.Offset],
    end: org.apache.spark.sql.execution.streaming.Offset
  ): DataFrame = {
    val startIdx = start.map(_.asInstanceOf[SeqOffset].value).getOrElse(0)
    val endIdx = end.asInstanceOf[SeqOffset].value
    // 截取当前批次的数据
    val batchData = dataSeq.slice(startIdx, endIdx)
    // 更新Offset
    currentOffset = endIdx
    // 转为DataFrame返回
    sqlContext.createDataFrame(batchData.map(Tuple1(_))).toDF("content")
  }

  override def stop(): Unit = {}

  // 注册数据源的短名称，用于readStream.format()调用
  override def shortName(): String = "seq-stream"
}

// 注册自定义数据源
object SeqStreamSource {
  def register(): Unit = {
    org.apache.spark.sql.execution.datasources.DataSource.register(classOf[SeqStreamSource])
  }
}

使用自定义数据源 #

// 注册自定义数据源
SeqStreamSource.register()

// 读取流式DataFrame
val streamingDF = spark.readStream
  .format("seq-stream")
  .option("data", rawSeq.mkString(",")) // 传入原始Seq数据
  .load()

// 启动流查询
val streamQuery = streamingDF.writeStream
  .outputMode("append")
  .format("console")
  .start()

streamQuery.awaitTermination()

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