Unable to log metrics to table/driver using spark StreamingQueryListener

[harry546]

Hi Everyone,

Hope everyone is enjoying their holiday time.

I have spark streaming job (copy activity from location A to location B) and a Query Listener and attached to the streaming job.

As mentioned over the attachments I can see numInputRows gets logged to driver logs. But I can't write the finaldf to delta table and even can't log the dataframe to driver logs when I do "show".

I remember the same piece code used to work on earlier versions may be at spark 3.* which was an year back.

Please let me know if I am doing something wrong or suggest me if changes required.

Thanks

Harish Koduru

[Kaniz_Fatma]

Hi @harry546, It sounds like you’re dealing with a Spark Streaming job that copies data from location A to location B, and you’ve attached a Query Listener to it.

 You’re encountering issues related to writing the finaldf to a Delta table and logging the DataFrame when using the show method.

Let’s break down the problem and explore potential solutions:

Writing to Delta Table:

• If you’re unable to write finaldf to a Delta table, consider the following:
  • Check Permissions: Ensure that the user running the Spark job has the necessary permissions to write to the target Delta table location.
  • Table Exists: Verify that the Delta table exists at the specified location. If not, create it using the appropriate schema.
  • Check Data Types: Confirm that the data types of columns in finaldf match the schema of the Delta table.
  • Checkpoint Location: If you’re using checkpoints, ensure that the checkpoint location is accessible and writable.
  • Error Logs: Check the driver logs for any error messages related to writing to the Delta table.

Logging DataFrame:

• If you’re unable to log the DataFrame using show, consider the following:
  • Driver Logs: Check the driver logs for any exceptions or issues related to the show operation.
  • Data Size: If finaldf contains a large amount of data, calling show might be resource-intensive. Consider limiting the number of rows displayed (e.g., finaldf.show(10)).
  • Memory Constraints: Ensure that the driver has sufficient memory to handle the DataFrame display.
  • Column Truncation: If the DataFrame has many columns, some columns may be truncated in the output. You can adjust the display settings using spark.conf.set("spark.sql.repl.eagerEval.enabled", True).

Spark Version Compatibility:

• You mentioned that the code used to work on earlier versions (around Spark 3.*). It’s possible that there are version-specific changes or compatibility issues.
• Consider checking the release notes for any relevant changes between your current Spark version and the version where the code worked.

Migration Strategy:

• If you need to change locations and table names while preserving data and checkpoints, consider the following approach:
  • Read from the old location and write to the new location using Spark Streaming.
  • Create the new Delta table at the new location.
  • Once the migration is complete, update your streaming jobs to read from the new location and use the new table names.

If you encounter any specific errors or need further assistance, feel free to provide additional details, and I’ll be happy to assist! 🚀