pyspark structured streaming not updating query metrics with query.lastProgress or other standard metrics

user_7164961 0
Viewed 616

I'm trying to add logging to my pyspark structured streaming application in order see the progress and statistics about each micro-batch processed. The writestream method uses a foreach writer to write the rows from the dataframe to a postgres database. I'm using .lastProgress and other standard metrics offered by pyspark for logging. The writestream method and my attempt to log are shown below.

query_1 = eventsDF \
    .writeStream \
    .foreach(writer) \
    .outputMode("append") \
    .option("checkpointLocation", "/tmp/checkpoint_a/") \
    .trigger(processingTime="5 seconds") \
    .start()


query_progress =  query_1.lastProgress
print("progress ", query_progress)
print("status ", query_1.status)
print("active ", query_1.isActive)

query_1.awaitTermination()

The results from my first loop are:

progress  None
status  {'message': 'Initializing sources', 'isDataAvailable': False, 'isTriggerActive': False}
active  True

However, further batches processed as event data arrives do not result in any more logging messages. I would expect that the logging messages would be emitted after each micro-batch processed in the streaming job.

I appreciate any advice or guidance. Thanks.

spark-structured-streaming pyspark apache-spark
1 Answers

All code between the start and the awaitTermination get executed only once. Only the code between load and start get executed contiuously on every query trigger.

According to the book "Spark - The definitive Guide" this way of monitoring is meant to run inside your application. However, for standalone application, you usually do not have a shell attached to run arbitrary code. In the book they recommend to "expose [the query] status by implementing a monitoring server, such as a small HTTP server that listens on a port and returns query.status when it gets a request.

Therefore, you need to create a dedicated runable thread that frequently calls the monitoring API of the query. I am really not familiar with Python, but it will essentially look like below:

# import the threading module 
import threading  
  
class thread(threading.Thread):  
    def __init__(self, query):  
        threading.Thread.__init__(self)  
        self.query = query  
  
        # helper function to execute the threads 
    def run(self):  
        print("progress ", query.lastProgress);

Having done this, you need to place it between start and awaitTermination:

query_1 = eventsDF \
    [...]
    .start()

monitoring = thread(query_1)

query_1.awaitTermination()

In stead of a dedicated thread you could also loop over the status of the query using while(query_1.isActive).

For Scala users:

How to get progress of streaming query after awaitTermination?

user_122089100
Related