topic Re: Data processing metrics in Data Engineering

Data processing metrics

noorbasha534 — Sat, 01 Feb 2025 17:55:13 GMT

Dear all,

What are some proven ways of capturing data processing metrics (number of rows processed/updated/inserted, number of micro-batches etc etc) in a PySpark/SQL code based notebook irrespective of the fact it uses auto-loader, structured streaming, DLT etc.

At the moment, we do profile the tables using "desc history" command and capture these as a reactive step. But, I like to achieve the same in real-time, and prepare an operation dashboard so it is useful for operators to pro-actively find those tables that have high latency on a particular day ((could be sudden increase in volume of records impacting a whole bunch of other tables))

Appreciate your thoughts.

Br,

Noor.

Re: Data processing metrics

Alberto_Umana — Sun, 02 Feb 2025 02:37:27 GMT

Hi @noorbasha534,

You can use the StreamingQueryListener interface to capture metrics like the number of input rows, processing time, and batch duration. This can be integrated into your PySpark code to log these metrics in real-time.

Example:

from pyspark.sql.streaming import StreamingQueryListener

class MyListener(StreamingQueryListener):
    def onQueryStarted(self, event):
        print(f"Query started: {event.id}")

    def onQueryProgress(self, event):
        print(f"Query made progress: {event.progress}")

    def onQueryTerminated(self, event):
        print(f"Query terminated: {event.id}")

spark.streams.addListener(MyListener())

df = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "parquet") \
    .load("s3://your-bucket/path")

query = df.writeStream.format("delta") \
    .option("checkpointLocation", "s3://your-bucket/checkpoints") \
    .start("s3://your-bucket/output")

Re: Data processing metrics

noorbasha534 — Tue, 25 Feb 2025 19:09:10 GMT

@Alberto_Umana thanks for the reply. without the current code change, is it possible to extract these metrics from logs?