DeepSeek Spark Structured Streaming 结构化流详解

稚辉君.MCA.P9.JAVA

256人浏览 · 2026-02-13 09:01:17

稚辉君.MCA.P9.JAVA · 2026-02-13 09:01:17 发布

Spark Structured Streaming 结构化流详解

一、概述

1.1 什么是Structured Streaming

Structured Streaming是Apache Spark 2.0引入的流处理引擎，建立在Spark SQL引擎之上。它将流处理抽象为无限增长的表，提供声明式API，支持端到端exactly-once语义。

1.2 核心特性

· 声明式API：使用DataFrame/DataSet API
· 事件时间处理：内置支持事件时间和窗口操作
· 端到端容错：基于检查点和预写日志
· 多种输出模式：Append, Update, Complete
· 与批处理统一：同一套API处理批和流

二、编程模型

2.1 无界表模型

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming._

// 创建SparkSession
val spark = SparkSession.builder
  .appName("StructuredStreamingExample")
  .master("local[*]")
  .getOrCreate()

// 读取流数据（模拟数据源）
val lines = spark.readStream
  .format("rate")
  .option("rowsPerSecond", 10)
  .load()

2.2 基本架构

输入源 → 无界表 → 查询 → 结果表 → 输出接收器
        ↑        ↓        ↓
     微批处理  持续查询  触发输出

三、输入源（Sources）

3.1 文件源

val fileStream = spark.readStream
  .format("json")  // 或 "csv", "parquet", "text"
  .schema(schema)  // 定义schema
  .option("path", "/path/to/files")
  .option("maxFilesPerTrigger", 10)  // 每次触发最大文件数
  .load()

3.2 Kafka源

val kafkaStream = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1,topic2")
  .option("startingOffsets", "earliest")  // latest, earliest, 或指定offsets
  .load()
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")

3.3 Socket源（测试用）

val socketStream = spark.readStream
  .format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load()

3.4 Rate源（测试用）

val rateStream = spark.readStream
  .format("rate")
  .option("rowsPerSecond", 100)  // 每秒生成的行数
  .option("rampUpTime", "5s")    // 逐渐增加到指定速率的时间
  .load()

四、数据处理

4.1 基本转换操作

// 类似批处理的DataFrame操作
val wordCounts = lines
  .select(explode(split(col("value"), " ")).as("word"))
  .groupBy("word")
  .count()

4.2 窗口操作（基于事件时间）

// 定义水印（处理延迟数据）
val windowedCounts = lines
  .withWatermark("timestamp", "10 minutes")  // 10分钟水印
  .groupBy(
    window(col("timestamp"), "10 minutes", "5 minutes"),  // 窗口：10分钟，滑动5分钟
    col("word")
  )
  .count()

4.3 水印（Watermarking）

// 处理延迟数据
val events = inputStream
  .select(
    col("data"),
    col("timestamp").cast("timestamp").as("eventTime")
  )
  .withWatermark("eventTime", "2 hours")  // 最多容忍2小时延迟

4.4 流-流Join

// 支持内连接、外连接
val joinStream = stream1
  .withWatermark("timestamp", "1 hour")
  .join(
    stream2.withWatermark("timestamp", "2 hours"),
    expr("""
      stream1.key = stream2.key AND
      stream1.timestamp >= stream2.timestamp AND
      stream1.timestamp <= stream2.timestamp + interval 1 hour
    """),
    joinType = "leftOuter"
  )

五、输出操作（Sinks）

5.1 输出模式

// Append模式（默认）：只添加新行
val query = wordCounts.writeStream
  .outputMode("append")
  .format("console")
  .start()

// Update模式：输出更新的行
val query = wordCounts.writeStream
  .outputMode("update")
  .format("console")
  .start()

// Complete模式：输出完整结果表
val query = wordCounts.writeStream
  .outputMode("complete")
  .format("console")
  .start()

5.2 输出接收器

// 1. Console Sink（调试用）
wordCounts.writeStream
  .outputMode("complete")
  .format("console")
  .option("truncate", false)
  .start()

// 2. File Sink
wordCounts.writeStream
  .outputMode("append")
  .format("parquet")
  .option("path", "/output/path")
  .option("checkpointLocation", "/checkpoint/path")
  .partitionBy("date")  // 按分区列写入
  .start()

// 3. Kafka Sink
wordCounts.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host:port")
  .option("topic", "output-topic")
  .option("checkpointLocation", "/checkpoint/path")
  .start()

// 4. Foreach Sink（自定义处理）
wordCounts.writeStream
  .foreach(new ForeachWriter[Row] {
    def open(partitionId: Long, epochId: Long): Boolean = {
      // 打开连接
      true
    }
    def process(row: Row): Unit = {
      // 处理每一行
    }
    def close(errorOrNull: Throwable): Unit = {
      // 关闭连接
    }
  })
  .start()

5.3 内存Sink

val query = wordCounts.writeStream
  .outputMode("complete")
  .format("memory")
  .queryName("wordCountsTable")  // 内存表名
  .start()

// 在Spark SQL中查询
spark.sql("SELECT * FROM wordCountsTable").show()

六、触发器和检查点

6.1 触发器

// 1. 默认触发器（尽可能快地处理）
.trigger(Trigger.ProcessingTime("0 seconds"))

// 2. 固定间隔触发器
.trigger(Trigger.ProcessingTime("1 minute"))

// 3. 一次性触发器
.trigger(Trigger.Once())

// 4. 连续触发器（实验性，低延迟）
.trigger(Trigger.Continuous("1 second"))

6.2 检查点

wordCounts.writeStream
  .outputMode("complete")
  .format("console")
  .option("checkpointLocation", "/path/to/checkpoint")
  // 恢复相关配置
  .option("failOnDataLoss", "false")  // 数据丢失时是否失败
  .start()

七、完整示例

7.1 实时词频统计

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

object StructuredStreamingWordCount {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("StructuredNetworkWordCount")
      .master("local[*]")
      .getOrCreate()
    
    import spark.implicits._
    
    // 从socket读取数据
    val lines = spark.readStream
      .format("socket")
      .option("host", "localhost")
      .option("port", 9999)
      .load()
    
    // 数据处理
    val words = lines.as[String]
      .flatMap(_.split(" "))
      .filter(_ != "")
    
    val wordCounts = words.groupBy("value").count()
    
    // 输出查询
    val query = wordCounts.writeStream
      .outputMode("complete")
      .format("console")
      .trigger(Trigger.ProcessingTime("5 seconds"))
      .option("checkpointLocation", "/tmp/checkpoint")
      .start()
    
    query.awaitTermination()
  }
}

7.2 实时异常检测

object RealTimeAnomalyDetection {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("RealTimeAnomalyDetection")
      .getOrCreate()
    
    // 读取Kafka流
    val kafkaDF = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "metrics")
      .load()
    
    // 解析JSON数据
    val schema = StructType(Seq(
      StructField("timestamp", TimestampType),
      StructField("service", StringType),
      StructField("metric", StringType),
      StructField("value", DoubleType)
    ))
    
    val metricsDF = kafkaDF
      .select(from_json(col("value").cast("string"), schema).as("data"))
      .select("data.*")
    
    // 检测异常（值大于阈值）
    val anomalies = metricsDF
      .withWatermark("timestamp", "5 minutes")
      .groupBy(
        window(col("timestamp"), "10 minutes", "5 minutes"),
        col("service"),
        col("metric")
      )
      .agg(avg("value").as("avg_value"), stddev("value").as("stddev"))
      .withColumn("z_score", 
        (col("value") - col("avg_value")) / col("stddev"))
      .filter(abs(col("z_score")) > 3.0)  // 3个标准差外的视为异常
    
    // 输出到Kafka
    val query = anomalies
      .select(to_json(struct("*")).as("value"))
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("topic", "anomalies")
      .option("checkpointLocation", "/tmp/anomaly-checkpoint")
      .start()
    
    query.awaitTermination()
  }
}

八、性能优化

8.1 配置优化

spark.conf.set("spark.sql.shuffle.partitions", "200")  // 调整shuffle分区数
spark.conf.set("spark.sql.streaming.stateStore.providerClass", 
  "org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider")

8.2 状态管理

// 设置状态超时
val aggregated = inputStream
  .groupBy(col("userId"))
  .agg(count("*").as("count"))
  .withTimeout("userId", "1 hour")  // 1小时后清理状态

8.3 使用MapGroupsWithState

// 自定义状态管理
val sessionUpdates = events
  .groupByKey(event => event.sessionId)
  .mapGroupsWithState[SessionInfo, SessionUpdate](
    GroupStateTimeout.ProcessingTimeTimeout) {
    
    case (sessionId: String, events: Iterator[Event], state: GroupState[SessionInfo]) =>
      // 处理逻辑
      if (state.hasTimedOut) {
        // 状态超时处理
      } else {
        // 更新状态
      }
  }

九、监控和管理

9.1 查询状态监控

val query = wordCounts.writeStream
  .format("console")
  .start()

// 获取查询信息
println(s"Query ID: ${query.id}")
println(s"Run ID: ${query.runId}")
println(s"Status: ${query.status}")
println(s"Last Progress: ${query.lastProgress}")

// 等待终止
query.awaitTermination()

9.2 监控指标

// 通过StreamingQueryListener监控
spark.streams.addListener(new StreamingQueryListener() {
  override def onQueryStarted(event: QueryStartedEvent): Unit = {
    println(s"Query started: ${event.name}")
  }
  override def onQueryProgress(event: QueryProgressEvent): Unit = {
    println(s"Processing stats: ${event.progress}")
  }
  override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {
    println(s"Query terminated: ${event.id}")
  }
})

十、故障恢复

10.1 从检查点恢复

// 自动从检查点恢复
val query = wordCounts.writeStream
  .outputMode("complete")
  .format("console")
  .option("checkpointLocation", "/path/to/checkpoint")
  .start()  // 如果检查点存在，会自动恢复