![\"Alt](\"https:\/\/devblogs.microsoft.com\/ise\/wp-content\/uploads\/sites\/55\/2025\/02\/spark_flow.png\")
<\/p>\n
Problem statement<\/h2>\n
In a traditional API enterprise job, OpenTelemetry<\/a> handles the start and end tracing for simple cases. However, in a Spark streaming job that runs for an extended period, potentially days, and processes millions of messages where calls might fail, its harder to keep track of failing batches.<\/p>\n StreamingQueryListener<\/a> is an excellent interface that is highly useful for instrumenting Spark jobs with OpenTelemetry. It’s a call-back for operations in batches that roughly equate to API requests, making it particularly beneficial for real-time monitoring of streaming data workloads.<\/p>\n In PySpark, the StreamingQueryListener listens for events related to StreamingQuery, providing methods such as onQueryStarted, onQueryProgress, onQueryIdle, and onQueryTerminated to manage different stages of a query\u2019s lifecycle.<\/p>\n Additionally, in Structured Streaming, you can define observable metrics<\/a>, which are essentially arbitrary aggregate functions applied to a query (DataFrame). These metrics can be monitored by attaching a listener to the Spark session. For further examples, such as the Kafka-to-Kafka StreamingQueryListener event, please refer here<\/a>.<\/p>\n The sample code below overrides Spark StreamingQueryListener functions onQueryStarted, onQueryTerminated, and onQueryProgress.<\/p>\n The following code shows how we utilized StreamingQueryListener to track the latency and performance of batch events.<\/p>\n In conclusion, the StreamingQueryListener class can be effectively utilized in the PySpark Streaming pipeline. This approach is also applicable to other Scala\/Java-supported libraries for PySpark.<\/p>\n Overall, the flexibility and adaptability of the StreamingQueryListener class make it a valuable tool for monitoring and optimizing streaming data workloads across various platforms and use cases.<\/p>\nSolution<\/h2>\n
Use cases<\/h3>\n
\n
Sample code<\/h3>\n
import datetime\r\n\r\nfrom opentelemetry import trace\r\nfrom pyspark.sql.streaming import StreamingQueryListener\r\n\r\ndef add_stream_progress_tracing(spark):\r\n listener = OpenTelemetryListener()\r\n spark.streams.addListener(listener)\r\n\r\nclass OpenTelemetryListener(StreamingQueryListener):\r\n def __init__(self):\r\n self.tracer = trace.get_tracer(__name__)\r\n\r\n def onQueryProgress(self, event):\r\n \"\"\"\r\n Called when there is some status update (ingestion rate updated, etc.)\r\n\r\n Parameters\r\n ----------\r\n event: :class:`pyspark.sql.streaming.listener.QueryProgressEvent`\r\n The properties are available as the same as Scala API.\r\n\r\n Notes\r\n -----\r\n This method is asynchronous. The status in\r\n :class:`pyspark.sql.streaming.StreamingQuery` returns the\r\n most recent status, regardless of when this method is called. The status\r\n of :class:`pyspark.sql.streaming.StreamingQuery`.\r\n may change before or when you process the event.\r\n For example, you may find :class:`StreamingQuery`\r\n terminates when processing `QueryProgressEvent`.\r\n \"\"\"\r\n progress = event.progress\r\n # epoch as int, multiply the result by 1_000_000_000 to convert to nanos\r\n start_time = int(\r\n datetime.datetime.fromisoformat(progress.timestamp).timestamp()\r\n * 1_000_000_000\r\n )\r\n batchDuration = progress.batchDuration\r\n # otel times are specified in nanos\r\n with self.tracer.start_span(\r\n name=\"process batch\",\r\n start_time=start_time,\r\n kind=trace.SpanKind.SERVER,\r\n ) as span:\r\n if progress.name:\r\n span.set_attribute(\"spark.streaming.query_name\", progress.name)\r\n span.set_attribute(\r\n \"spark.streaming.query_progress_id\", progress.id.__str__()\r\n )\r\n span.set_attribute(\"spark.streaming.batch_id\", progress.batchId)\r\n span.set_attribute(\"spark.streaming.num_input_rows\", progress.numInputRows)\r\n span.set_attribute(\r\n \"spark.streaming.input_rows_per_second\",\r\n progress.inputRowsPerSecond,\r\n )\r\n span.set_attribute(\r\n \"spark.streaming.processed_rows_per_second\",\r\n progress.processedRowsPerSecond,\r\n )\r\n # The time it takes to plan and execute the microbatch.\r\n span.set_attribute(\r\n \"spark.streaming.trigger_execution_time\",\r\n progress.durationMs[\"triggerExecution\"],\r\n )\r\n\r\n # Calculate the span end\r\n # batchDuration is in milliseconds, so multiply the result by 1_000_000 to convert to nanos\r\n end_time = start_time + (batchDuration * 1_000_000)\r\n span.end(end_time=end_time)\r\n\r\n def onQueryStarted(self, event):\r\n \"\"\"\r\n Called when a query is started.\r\n\r\n Parameters\r\n ----------\r\n event: :class:`pyspark.sql.streaming.listener.QueryStartedEvent`\r\n The properties are available as the same as Scala API.\r\n\r\n Notes\r\n -----\r\n This is called synchronously with\r\n meth:`pyspark.sql.streaming.DataStreamWriter.start`,\r\n that is, ``onQueryStart`` will be called on all listeners before\r\n ``DataStreamWriter.start()`` returns the corresponding\r\n :class:`pyspark.sql.streaming.StreamingQuery`.\r\n Do not block in this method as it will block your query.\r\n \"\"\"\r\n pass\r\n\r\n def onQueryTerminated(self, event):\r\n \"\"\"\r\n Called when a query is stopped, with or without error.\r\n\r\n Parameters\r\n ----------\r\n event: :class:`pyspark.sql.streaming.listener.QueryTerminatedEvent`\r\n Determine whether the query ended with an error and retrieve the corresponding error details from the StreamingQueryProgress object.\r\n The properties are available as the same as Scala API.\r\n \"\"\"\r\n with tracer.start_as_current_span(\"QueryTerminated\") as span:\r\n # Record the query ID as an attribute\r\n span.set_attribute(\"query.id\", str(event.id))\r\n if event.exception is not None:\r\n # Mark the span with error details if an exception occurred\r\n span.set_attribute(\"query.error\", True)\r\n span.set_attribute(\"query.error.message\", str(event.exception))\r\n print(f\"Query terminated with an error: {event.exception}\")\r\n else:\r\n span.set_attribute(\"query.error\", False)\r\n print(\"Query terminated successfully.\")<\/code><\/pre>\nSummary<\/h2>\n