Spark Structured Streaming三大限制咨询：性能、流计算、监控

Hey there! Let's dive into your questions about Structured Streaming— I’ve spent plenty of time migrating from DStreams to Structured Streaming, so I can share practical insights here.

Q1: Will multiple Sinks cause independent source connections and performance degradation? #

Great question, but you might be surprised by how Structured Streaming handles this by default:

• It does NOT create 3 independent source connections if you’re writing multiple outputs from the same input stream in a single job. Structured Streaming uses a shared input model— once it pulls data from Kafka (or any source), it replicates the processed data to all configured Sinks internally. This is way more efficient than DStreams' single-connection approach because the Spark optimizer can apply unified optimizations (like predicate pushdown) to the input once, instead of repeating work.
• The only scenario where you’d get multiple source connections is if you create three separate readStream jobs (each reading from Topic A and writing to one Sink). That’s a bad practice anyway— always consolidate multiple outputs into a single streaming query to leverage shared input processing.

Q2: How to compute joins between two streaming datasets (Topic A and Topic B)? #

First, let’s clarify: Structured Streaming does support stream-stream joins, but with some guardrails to avoid unbounded state growth. Here’s how to handle it based on your use case:

• Stream-stream joins with time constraints (most common)：
  You need to define watermarks for both streams to manage late data, and specify a time-based join condition. For example, joining user events from Topic A and Topic B within a 1-hour window:

  import org.apache.spark.sql.functions.expr
  
  // Define watermarks to control state retention
  val streamA = spark.readStream
    .format("kafka")
    .load()
    .selectExpr("cast(value as string) as data")
    .withColumn("user_id", expr("get_json_object(data, '$.user_id')"))
    .withColumn("event_time", expr("to_timestamp(get_json_object(data, '$.event_time'))"))
    .withWatermark("event_time", "2 hours") // Allow 2 hours of late data
  
  val streamB = spark.readStream
    .format("kafka")
    .load()
    .selectExpr("cast(value as string) as data")
    .withColumn("user_id", expr("get_json_object(data, '$.user_id')"))
    .withColumn("event_time", expr("to_timestamp(get_json_object(data, '$.event_time'))"))
    .withWatermark("event_time", "2 hours")
  
  // Perform inner join with 1-hour time window
  val joinedStream = streamA.join(
    streamB,
    expr("streamA.user_id = streamB.user_id AND streamA.event_time BETWEEN streamB.event_time - interval 1 hour AND streamB.event_time"),
    joinType = "inner"
  )
  

• Stream-static join: If one of the streams doesn’t change frequently (or you can tolerate some latency), you can materialize Topic B into a static DataFrame (e.g., read from a parquet table updated by a separate batch job) and join it with Topic A’s stream. This avoids state management entirely.
• Unbounded stream-stream joins (not recommended): If you truly need a join without time constraints, Structured Streaming blocks this by default— it would lead to infinite state size as more data comes in. In this case, you’d need to manually manage state using flatMapGroupsWithState or mapGroupsWithState, but this requires careful state cleanup logic to avoid out-of-memory issues.

Q3: Why no dedicated Streaming tab in Spark UI, and how to monitor Structured Streaming? #

Structured Streaming is built on top of Spark SQL’s execution engine, so its metrics live in the SQL tab of the Spark UI instead of a separate Streaming tab. Here’s what you need to know:

• In the SQL tab, you’ll see your streaming query listed as a "continuous" query. Click into it to view metrics like processing rate (rows per second), latency, input rows, and state size.
• For advanced monitoring beyond the UI:
  • Use StreamingQuery.status() to get real-time status of your query (e.g., active, idle, processing progress).
  • Implement StreamingQueryListener to listen to query events (start, progress, termination, errors) and push these metrics to your monitoring system (like Prometheus, Grafana, or a custom dashboard).
  • Configure Spark’s built-in metrics system (via metrics.properties) to expose JVM, Spark, and Structured Streaming metrics to external monitoring tools.

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