如何在Apache Spark中跟踪防火墙日志的共享状态？

Great question! Tracking paired start/stop events per UniqueID is a super common task in Spark, whether you're working with batch historical logs or real-time streaming data. Let's walk through the best approaches for both scenarios, with a focus on handling the UniqueID state correctly.

Batch Processing (Historical Logs) #

If you're working with static, already-collected firewall logs, you can leverage Spark's relational operations to match start and stop events. Here are two reliable methods:

Method 1: Split and Join (for 1:1 Event Pairs) #

If each start has exactly one corresponding stop per UniqueID, split the data into start/stop datasets and join on UniqueID. For safety, add a check to ensure stop timestamps are after start timestamps:

// Assume rawLogs is your input DataFrame with columns: UniqueID, start_or_stop, timestamp
val startDF = rawLogs
  .filter($"start_or_stop" === "start")
  .select($"UniqueID", $"timestamp".alias("start_time"))

val stopDF = rawLogs
  .filter($"start_or_stop" === "stop")
  .select($"UniqueID", $"timestamp".alias("stop_time"))

// Join and filter out invalid pairs (stop before start)
val resultDF = startDF.join(stopDF, Seq("UniqueID"), "inner")
  .filter($"stop_time" > $"start_time")

Method 2: Windowed Row Numbering (for Multiple Pairs per ID) #

If a UniqueID can have multiple start/stop sessions (e.g., repeated connections), use window functions to number events per ID, then match odd-numbered starts with even-numbered stops:

import org.apache.spark.sql.expressions.Window

// Define a window partitioned by UniqueID, ordered by timestamp
val eventWindow = Window.partitionBy("UniqueID").orderBy("timestamp")

// Add a sequence number to each event for the ID
val numberedEvents = rawLogs
  .withColumn("event_seq", row_number().over(eventWindow))

// Split into numbered start/stop datasets
val numberedStarts = numberedEvents
  .filter($"start_or_stop" === "start")
  .select($"UniqueID", $"timestamp".alias("start_time"), $"event_seq")

val numberedStops = numberedEvents
  .filter($"start_or_stop" === "stop")
  .select($"UniqueID", $"timestamp".alias("stop_time"), $"event_seq")

// Join on ID and sequence number to pair matching start/stop events
val sessionResults = numberedStarts.join(numberedStops, Seq("UniqueID", "event_seq"), "inner")
  .drop("event_seq")

Streaming Processing (Real-Time Logs) #

For live firewall streams, you need to maintain state for each UniqueID to track pending start events until their corresponding stop arrives. Spark Structured Streaming provides flatMapGroupsWithState for exactly this kind of stateful processing.

Step 1: Define State and Output Classes #

First, create case classes to represent the state we need to track (start timestamps) and our final output:

// Tracks the pending start time for a UniqueID
case class PendingStart(startTime: Long)

// Final output structure: (UniqueID, start_time, stop_time)
case class SessionResult(uniqueID: String, startTime: Long, stopTime: Long)

Step 2: Implement State Processing Logic #

Write a function that processes events per UniqueID, manages state, and emits results when a stop event is matched:

import org.apache.spark.sql.streaming.{GroupState, GroupStateTimeout}

def processSessionEvents(
    uniqueID: String,
    events: Iterator[(String, Long)], // Tuple of (start_or_stop, timestamp)
    state: GroupState[PendingStart]
): Iterator[SessionResult] = {

  val results = scala.collection.mutable.ArrayBuffer[SessionResult]()

  events.foreach { case (eventType, timestamp) =>
    eventType match {
      case "start" =>
        // If no existing state, record the start time; handle duplicates as needed
        if (!state.exists) {
          state.update(PendingStart(timestamp))
          // Optional: Set a timeout to clean up stale start events (e.g., 1 hour)
          state.setTimeoutDuration("1 hour")
        } else {
          // Handle duplicate starts (e.g., log a warning, or update to latest timestamp)
          println(s"Warning: Duplicate start event for ID $uniqueID at $timestamp")
        }

      case "stop" =>
        // If there's a pending start, emit the result and clear the state
        if (state.exists) {
          val pendingStart = state.get
          results += SessionResult(uniqueID, pendingStart.startTime, timestamp)
          state.remove()
        } else {
          // Handle orphaned stop events (no matching start)
          println(s"Warning: Orphaned stop event for ID $uniqueID at $timestamp")
        }
    }
  }

  // Clean up timed-out start events (optional: emit as incomplete sessions)
  if (state.hasTimedOut) {
    val staleStart = state.get
    println(s"Timeout: No stop event for ID $uniqueID (started at ${staleStart.startTime})")
    state.remove()
  }

  results.iterator
}

Step 3: Apply to Streaming Data #

Wire up the state function to your stream, and add watermarking to handle out-of-order events:

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

// Assume rawStream is your input streaming DataFrame
val processedStream = rawStream
  // Add watermark to handle up to 10 minutes of out-of-order events
  .withWatermark("timestamp", "10 minutes")
  .select($"UniqueID", $"start_or_stop", $"timestamp")
  // Group events by UniqueID
  .groupByKey(_.UniqueID)
  // Apply state processing with processing-time timeouts
  .flatMapGroupsWithState(GroupStateTimeout.ProcessingTimeTimeout)(processSessionEvents)

// Output the results (e.g., to console, Kafka, or a data lake)
processedStream.writeStream
  .outputMode("append")
  .format("console")
  .start()
  .awaitTermination()

Key Considerations #

• Event Order: Use watermarking to handle out-of-order events in streams, ensuring stop events don't arrive before their corresponding start.
• State Management: Spark persists state to ensure fault tolerance (configurable via spark.sql.streaming.checkpointLocation).
• Edge Cases: Always handle orphaned stops, duplicate starts, and timed-out starts to avoid stale state and incorrect results.

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