Structured Streaming性能监控与对比及QueryProgress输出技术问询

Great questions—let’s tackle each one with practical, actionable advice since Structured Streaming monitoring and cross-framework performance comparisons are key when building or optimizing streaming pipelines.

Spark UI is great for real-time debugging, but for long-term monitoring, alerting, and granular insights, these options are more robust:

• Enhanced Custom StreamingQueryListener: You already started with this approach, but you can extend it to capture deeper metrics like per-operator processing time, state storage size (for stateful operations like aggregations/windows), and end-to-end event time delay. Push these metrics to a time-series monitoring system like Prometheus, then visualize with Grafana. For example, extract durationMs.triggerExecution (total time per trigger) and stateOperators[0].numRows (state size) from the QueryProgress object, then expose them via Prometheus’ Java client for real-time dashboards and alerting.
• Spark Metrics System Integration: Spark’s built-in metrics system supports sinks like Prometheus, Graphite, or JMX. Structured Streaming automatically exposes core metrics like streaming.inputRowsPerSecond, streaming.processedRowsPerSecond, and streaming.stateStoreSize. Configure these in spark/conf/metrics.properties to push metrics directly to your monitoring stack—this avoids custom code overhead and gives you out-of-the-box, cluster-wide visibility.
• Structured Logging with ELK Stack: Configure your logging framework (Logback/Log4j2) to output QueryProgress as structured JSON logs. Use Logstash to ingest these logs into Elasticsearch, then build custom dashboards in Kibana to analyze historical performance, troubleshoot latency spikes, or track query trends over time. This is especially useful for auditing and long-term performance analysis.

Both approaches leverage custom StreamingQueryListener, but here’s how to implement them reliably:

Output to Files #

• Custom Listener with Distributed Storage: In your listener’s onQueryProgress method, convert the progress object to JSON using progress.json() and write it to a distributed file system (HDFS, S3, ADLS). Use thread-safe writing utilities (like Apache Commons IO’s FileUtils.writeStringToFile in append mode) or split files by time (e.g., hourly files) to avoid oversized logs. For distributed clusters, ensure all worker nodes write to a shared storage location so logs are centralized.
• Log Framework Filtering: Alternatively, adjust your log configuration to capture only QueryProgress logs. Set the log level for org.apache.spark.sql.execution.streaming.StreamingQueryListener to INFO, then add a dedicated appender in Logback/Log4j2 to route these logs to a specific file. This requires zero custom code and works seamlessly with existing logging pipelines.

Output to Kafka #

• Asynchronous Kafka Producer in Listener: Create a singleton Kafka Producer (to avoid resource bloat) in your listener, then send the JSON-serialized QueryProgress to a dedicated Kafka topic. Wrap the send operation in an async thread pool to avoid blocking the query’s trigger execution. Example snippet (Scala):

  import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
  import java.util.Properties
  
  class KafkaProgressListener(topic: String) extends StreamingQueryListener {
    private val kafkaProps = new Properties()
    kafkaProps.put("bootstrap.servers", "kafka-broker:9092")
    kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    private val producer = new KafkaProducer[String, String](kafkaProps)
  
    override def onQueryProgress(event: QueryProgressEvent): Unit = {
      val progressJson = event.progress.json()
      val record = new ProducerRecord[String, String](topic, progressJson)
      // Send asynchronously to avoid blocking the query
      producer.send(record, (metadata, exception) => {
        if (exception != null) {
          // Handle failure (e.g., log to dead-letter queue)
          exception.printStackTrace()
        }
      })
    }
  
    // Don't forget to close the producer when the query stops
    override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {
      producer.close()
    }
  }
  

• Error Handling: Add retry logic or a dead-letter topic for failed messages to ensure no progress data is lost. Avoid heavy sync operations in the listener—async sending keeps your query’s performance unaffected.

To get fair, actionable comparisons, focus on controlled testing and aligned metrics:

• Control the Test Environment: Use identical cluster configurations (nodes, CPU, memory, network) and input sources. For example, generate a fixed-rate stream of test data via Kafka, or repeat a large file dataset. Match key configurations: set Spark Streaming’s batch interval equal to Structured Streaming’s trigger interval, use the same state storage (e.g., RocksDB for both), and align parallelism levels.
• Collect Aligned Metrics:
  • Input/Output Records: For Spark Streaming, use StreamingListener.onBatchCompleted to get batchInfo.numRecords (input) and track output counts via DStream.count(). For Structured Streaming, use QueryProgress.numInputRows and numOutputRows. Push all metrics to the same monitoring system (Prometheus/InfluxDB) for side-by-side comparison.
  • Speed & Latency: Track Spark Streaming’s batch processing delay (batchInfo.processingDelay) and end-to-end latency. For Structured Streaming, use progress.durationMs.triggerExecution (processing time per trigger) and eventTimeDelay (lag between event time and processing time). Measure end-to-end latency by tagging test data with timestamps and tracking when they reach the output sink.
• Side-by-Side Pipeline Testing: Deploy identical business logic pipelines (e.g., windowed aggregations, joins) in both frameworks, processing the same input stream. Run tests for hours (not just minutes) to capture stability trends—look for memory leaks, state growth, and recovery time after failures (e.g., killing a worker node).
• Benchmark with Standard Tools: Use Spark’s built-in Benchmark class to write repeatable performance tests, or use Apache JMeter to generate consistent load. Compare throughput (records per second) and resource utilization (CPU, memory) across both frameworks to identify bottlenecks.

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