監控流查詢結構化數據磚

類MyListener(StreamingQueryListener):defonQueryStarted(自我,事件):”“”啟動時調用查詢。參數- - - - - - - - - - -事件::類:“pyspark.sql.streaming.listener.QueryStartedEvent”可用的屬性是一樣的Scala API。筆記- - - - - -這就是所謂的同步甲:“pyspark.sql.streaming.DataStreamWriter.start”,即“onQueryStart“將呼籲所有聽眾“DataStreamWriter.start() ' '返回對應的類:“pyspark.sql.streaming.StreamingQuery”。不阻塞在這個方法,因為它會阻止您的查詢。”“”通過defonQueryProgress(自我,事件):”“”當有一些狀態更新(攝入率更新等)。參數- - - - - - - - - - -事件::類:“pyspark.sql.streaming.listener.QueryProgressEvent”可用的屬性是一樣的Scala API。筆記- - - - - -這個方法是異步的。的狀態:類:“pyspark.sql.streaming。StreamingQuery”返回最近的狀態,無論何時調用此方法。狀態:類的:“pyspark.sql.streaming.StreamingQuery”。或者當你處理事件之前可能會改變。例如,你可能會發現:類:“StreamingQuery”終止在處理“QueryProgressEvent”。”“”通過defonQueryTerminated(自我,事件):”“”當停止查詢,有或沒有錯誤。參數- - - - - - - - - - -事件::類:“pyspark.sql.streaming.listener.QueryTerminatedEvent”可用的屬性是一樣的Scala API。”“”通過my_listener=MyListener()

#觀察指標observed_df=df。觀察(“指標”,數(點燃(1))。作為(“問”),數(上校(“錯誤”))。作為(“畸形”))observed_df。writeStream。格式(“…”)。開始()#定義我的聽眾。類MyListener(StreamingQueryListener):defonQueryStarted(自我,事件):打印(f“‘{事件。的名字}”({事件。id})開始!”)defonQueryProgress(自我,事件):行=事件。進步。observedMetrics。得到(“指標”)如果行是不沒有一個:如果行。畸形的/行。問>0.5:打印(“警告!哎喲!有太多的畸形”f“記錄{行。畸形的}的{行。問}!”)其他的:打印(f”{行。問}行處理!”)defonQueryTerminated(自我,事件):打印(f”{事件。id}終止了!”)#添加我的偵聽器。火花。流。addListener(MyListener())