article Unified Logging for Databricks Notebooks and ADF with Azure Log Analytics in Technical Blog

article Unified Logging for Databricks Notebooks and ADF with Azure Log Analytics in Technical Blog https://community.databricks.com/t5/technical-blog/unified-logging-for-databricks-notebooks-and-adf-with-azure-log/ba-p/64717 <P><LI-TOC indent="15" liststyle="disc" maxheadinglevel="2"></LI-TOC></P> <H1>Introduction</H1> <H2>Logging in Azure Data Factory and Databricks Notebooks</H2> <P>Today we are looking at logging for Azure Data Factory (ADF) and Databricks Notebooks. Both of these tools separately have great solutions for logging, but they don't mesh well:</P> <UL> <LI>ADF does not persist logs indefinitely unless you specifically ask it to by <A href="https://learn.microsoft.com/en-us/azure/data-factory/monitor-configure-diagnostics" target="_self">configuring diagnostic logging</A>. You can send it to a <FONT face="arial,helvetica,sans-serif">storage location</FONT>, Log Analytics, or Event Hubs. Otherwise you can only see what happened from within ADF studio until the runs expire after 45 days.</LI> <LI>Databricks Notebooks just run code, so unless you set up <A href="https://learn.microsoft.com/en-us/azure/databricks/compute/configure#--compute-log-delivery" target="_self">compute log delivery</A> at the compute level (not very helpful in this case) which will export the <STRONG><FONT face="courier new,courier">stdout</FONT></STRONG>, <STRONG><FONT face="courier new,courier">stderr</FONT></STRONG>, and Spark logs, you will need some sort for custom logging solution based off <FONT face="courier new,courier"><STRONG><A href="https://logging.apache.org/log4j/2.x/" target="_self">log4j</A></STRONG></FONT> or Python's <STRONG><A href="https://docs.python.org/3/library/logging.html" target="_self"><FONT face="courier new,courier">logging</FONT></A></STRONG> package. </LI> </UL> <H2>Solution Requirements</H2> <P>The best solution has the following requirements:</P> <UL> <LI>Persist all logs to the same location</LI> <LI>Persist all logs indefinitely</LI> <LI>Ensure the ADF logs for a given notebook and pipeline can be tied to the notebook logs</LI> <LI>Simple to set up and maintain</LI> </UL> <H1>Proposed Solution</H1> <P>The approach laid out in this article is to use ADF's native integration with Azure Log Analytics and then create a custom logging package using Python to send logs from Databricks Notebooks to Azure Log Analytics.</P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="Logging Overview.png" style="width: 695px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/6828iFC6FE04F0B7B078A/image-dimensions/695x534?v=v2" width="695" height="534" role="button" title="Logging Overview.png" alt="Logging Overview.png" /></span></P> <P>This will enable us to note only keep track of the pipeline and notebook runs, but also link the logs in a given notebook to the corresponding logs for that notebook activity run in ADF.</P> <H2>Custom Logging Package</H2> <P>In order to send logs from Databricks Notebooks to Log Analytics we need client code running on Databricks which meets following requirements:</P> <UL> <LI>Must run in Databricks notebooks</LI> <LI>Must be configurable for the Log Analytics instance details: <UL> <LI> <DIV><FONT face="courier new,courier"><STRONG>LOGGING_WORKSPACE_ID:</STRONG><FONT face="arial,helvetica,sans-serif"> the Log Analytics workspace ID</FONT></FONT></DIV> </LI> <LI> <DIV><FONT face="courier new,courier"><SPAN><STRONG>LOGGING_WORKSPACE_KEY:</STRONG><FONT face="arial,helvetica,sans-serif"> the Log Analytics workspace key</FONT></SPAN></FONT></DIV> </LI> <LI> <DIV><FONT face="courier new,courier"><SPAN><STRONG>LOG_TABLE_NAME:</STRONG><FONT face="arial,helvetica,sans-serif"> the table name logs will be pushed to in Log Analytics</FONT></SPAN></FONT></DIV> </LI> </UL> </LI> <LI><FONT face="courier new,courier"><SPAN><FONT face="arial,helvetica,sans-serif">Must be configurable within a notebook for the pipeline and activity information so that we can link the notebook logs with the ADF logs for that notebook activity:<BR /></FONT></SPAN></FONT> <UL> <LI><FONT face="courier new,courier"><SPAN><FONT face="arial,helvetica,sans-serif">pipeline run ID</FONT></SPAN></FONT></LI> <LI><FONT face="courier new,courier"><SPAN><FONT face="arial,helvetica,sans-serif">pipeline name</FONT></SPAN></FONT></LI> <LI><FONT face="courier new,courier"><SPAN><FONT face="arial,helvetica,sans-serif">activity name</FONT></SPAN></FONT></LI> </UL> </LI> </UL> <P><FONT face="courier new,courier"><SPAN><FONT face="arial,helvetica,sans-serif">Prototype code that achieves this is provided in this </FONT></SPAN></FONT><A href="https://github.com/sdanielzafar/databricks_loganalytics" target="_self">Github Repo</A>. Let's talk about the primary class, which is in <STRONG><FONT face="courier new,courier">log_analytics.py</FONT></STRONG>:</P> <P> </P> <LI-CODE lang="markup">from databricks_loganalytics.base_logging import log_console_output from pyspark.sql import SparkSession, DataFrame from databricks.sdk.runtime import * from time import sleep import os global workspace_id global workspace_key workspace_id: str = os.environ.get('LOGGING_WORKSPACE_ID') workspace_key: str = os.environ.get('LOGGING_WORKSPACE_KEY') spark: SparkSession = SparkSession.builder.getOrCreate() class notebook_logger: def __init__(self, pipeline_run_id: str, pipeline_name: str, activity_name: str): self.pipeline_run_id: str = pipeline_run_id self.pipeline_name: str = pipeline_name self.activity_name: str = activity_name def log_info(self, *messages) -> None: message: str = " ".join([str(m) for m in messages]) dbx_body: dict[str, str] = [ { "Message": message, "pipelineRunId_g": self.pipeline_run_id, "pipeline": self.pipeline_name, "activity": self.activity_name, } ] res: str = log_console_output( workspace_id=workspace_id, workspace_key=workspace_key, message=dbx_body, ) print(message) if res != 'Accepted': print(res) def log_notebook_output(self, output: dict): def log_cmd_output(output): if isinstance(output, DataFrame): try: out = output.collect() if out: outDict = out[0].asDict() match len(outDict): case 2: operation: str = "INSERT" case 4: operation: str = "MERGE" case 1: operation: str = "DELETE or UPDATE" case _: operation: str = "UKNOWN" if len(out) == 1: message: str = f"{operation}: " + ", ".join([f"{op}: {str(cnt)}" for op, cnt in outDict.items()]) self.log_info(message) sleep(0.1) except Exception as e: # Handle any exceptions that may occur print(f"WARNING: {output} not logged") elif isinstance(output, str): self.log_info(output) [log_cmd_output(values) for _, values in output.items()] def log_inserted_count(self, table): message: str = f"INSERTED {spark.read.table(table).count()}" self.log_info(message) </LI-CODE> <P> </P> <P>Here we define the class <STRONG><FONT face="courier new,courier">notebook_logger</FONT></STRONG>, in which the constructor takes input arguments which can be used to link logs to their respective ADF activity runs. The best way to approach this is to pass in values for these as notebook widgets (ADF parameters), where the <STRONG><FONT face="courier new,courier">activity_name</FONT></STRONG> matches the name given to the ADF notebook activity. More on this later.</P> <P>The three methods provided by <STRONG><FONT face="courier new,courier">notebook_logger</FONT></STRONG><FONT face="courier new,courier"><FONT face="arial,helvetica,sans-serif"> offer different ways to send logs to Log Analytics:</FONT></FONT></P> <UL> <LI><FONT face="courier new,courier"><FONT face="arial,helvetica,sans-serif"><FONT face="courier new,courier"><STRONG>log_info</STRONG></FONT>: passes a string to Log Analytics. Supports multiple arguments like python's <STRONG><FONT face="courier new,courier">print</FONT></STRONG> does.</FONT></FONT></LI> <LI><FONT face="courier new,courier"><FONT face="courier new,courier"><FONT face="arial,helvetica,sans-serif"><STRONG><FONT face="courier new,courier">log_notebook_output</FONT></STRONG>: intended to be run at the bottom of the notebook. if called at the bottom of the notebook like so:<BR /></FONT></FONT></FONT><LI-CODE lang="markup">logger.log_notebook_output(Out)</LI-CODE><FONT face="courier new,courier"><FONT face="arial,helvetica,sans-serif">where <FONT face="courier new,courier"><STRONG>Out</STRONG></FONT> is a variable Databricks Notebooks puts ipython <FONT face="courier new,courier"><STRONG>stdout</STRONG></FONT>, then any ipython <STRONG>stdout</STRONG> will be sent to <STRONG><FONT face="courier new,courier">log_info</FONT></STRONG>. This is really intended to send off the results of <FONT face="courier new,courier"><STRONG>MERGE</STRONG></FONT>, <STRONG><FONT face="courier new,courier">UPDATE</FONT></STRONG>, <STRONG><FONT face="courier new,courier">INSERT</FONT></STRONG>, and <FONT face="courier new,courier"><STRONG>DELETE</STRONG></FONT> operations.</FONT></FONT></LI> <LI><FONT face="courier new,courier"><FONT face="arial,helvetica,sans-serif"><STRONG><FONT face="courier new,courier">log_inserted_count</FONT></STRONG>: a convenience method to be used after an <STRONG><FONT face="courier new,courier">INSERT</FONT>.</STRONG></FONT></FONT></LI> </UL> <H1>Set Up</H1> <H2>Prerequisites</H2> <UL> <LI>A Log Analytics workspace accessible by ADF and Azure Databricks.</LI> <LI>The Workspace ID and the Workspace key for this Log Analytics workspace.</LI> <LI>Permissions to set environment variables on your Databricks cluster.</LI> </UL> <H2>Steps</H2> <OL> <LI>Integrate ADF and Log Analytics by <A href="https://learn.microsoft.com/en-us/azure/data-factory/monitor-configure-diagnostics" target="_self">configuring diagnostic logging</A> in your ADF instance. You will set the Log Analytics workspace. Make sure to check at least <STRONG>ActivityRuns</STRONG>, <STRONG>PipelineRuns</STRONG>, and <STRONG>TriggerRuns</STRONG>.</LI> <LI>Test that is working by running a test pipeline and then executing the following query in Log Analytics:<BR /><LI-CODE lang="markup">AzureDiagnostics | where ResourceId contains "<your-adf-resource-id>"</LI-CODE></LI> <LI>Add the custom logger to Databricks: <OL> <LI>Add the custom logger to your Databricks cluster using the cluster UI. <OL> <LI>Put the <A href="https://github.com/sdanielzafar/databricks_loganalytics/tree/main/dist" target="_self">.whl file here</A> on DBFS</LI> <LI>Install the .whl file on your Databricks cluster (<A href="https://docs.databricks.com/en/libraries/cluster-libraries.html#install-a-library-on-a-cluster" target="_self">reference</A>)</LI> <LI>Set environment variables on your Databricks cluster (<A href="https://docs.databricks.com/en/compute/configure.html#environment-variables" target="_self">reference</A>). The <STRONG><FONT face="courier new,courier">LOG_TABLE_NAME</FONT></STRONG> can be anything you prefer.<BR /><LI-CODE lang="markup">LOGGING_WORKSPACE_ID="<your-workspace-id>" LOGGING_WORKSPACE_KEY="<your-workspace-key>" LOG_TABLE_NAME="dbrx_notebook_logs"</LI-CODE></LI> </OL> </LI> <LI>(Alternatively) If you don't want to deal with whl files and especially if you are using Git Folders (formerly Databricks Repos), you can add the code in <A href="https://github.com/sdanielzafar/databricks_loganalytics/tree/main/databricks_loganalytics" target="_self">this directory</A> to your Databricks workspace.</LI> <LI>Ensure you can import the code by running this Python code in a notebook:<BR /><LI-CODE lang="markup">from databricks_loganalytics.log_analytics import notebook_logger</LI-CODE></LI> <LI>To ensure everything works, run this code in a Databricks Notebook:<BR /><LI-CODE lang="markup">from databricks_loganalytics.log_analytics import notebook_logger logger = notebook_logger("test", "test", "test") logger.log_info("This is a test message")</LI-CODE></LI> <LI>If everything is working properly you should be able to find this message in Log Analytics using this Kusto query (we set this from the environment variable above):<BR /><LI-CODE lang="markup">dbrx_notebook_logs</LI-CODE> <P>If that doesn't work, you can also try "<STRONG><FONT face="courier new,courier">dbrx_notebook_logs_CL</FONT></STRONG>".</P> </LI> </OL> </LI> </OL> <H1>Usage</H1> <P>At this point, if you've gotten the tests working you probably have a good idea of what's going on. Let's clear up any uncertainty with a real example. If you are not familiar with passing parameters between ADF and Databricks notebooks, or rusty, you can <A href="https://learn.microsoft.com/en-us/azure/data-factory/transform-data-using-databricks-notebook#create-a-pipeline" target="_self">review the docs here</A>.</P> <H2>ADF Activity Parameters</H2> <P>Let's say you have an ADF pipeline activity that looks like this:</P> <P><span class="lia-inline-image-display-wrapper lia-image-align-inline" image-alt="Dan_Z_3-1711642295309.png" style="width: 200px;"><img src="https://community.databricks.com/t5/image/serverpage/image-id/6852i798F40435E9621B4/image-size/small?v=v2&px=200" role="button" title="Dan_Z_3-1711642295309.png" alt="Dan_Z_3-1711642295309.png" /></span></P> <P>You will need to set the following parameters for the Notebook Activity:</P> <UL> <LI><STRONG><FONT face="courier new,courier">PIPELINE_RUN_ID</FONT></STRONG>: set to the variable <STRONG><FONT face="courier new,courier">@pipeline().RunId</FONT></STRONG><FONT face="courier new,courier"><FONT face="arial,helvetica,sans-serif"> (dynamic)</FONT></FONT></LI> <LI><FONT face="courier new,courier"><STRONG>PIPELINE_NAME</STRONG><FONT face="arial,helvetica,sans-serif">: </FONT></FONT>set to variable <STRONG><FONT face="courier new,courier">@pipeline().Pipeline</FONT></STRONG><FONT face="arial,helvetica,sans-serif"> (dynamic)</FONT></LI> <LI><STRONG><FONT face="courier new,courier">ACTIVITY_NAME</FONT></STRONG><FONT face="courier new,courier"><FONT face="arial,helvetica,sans-serif">: set to the string <STRONG><FONT face="courier new,courier">"LoadHistoricalDataIntoSmallTables".</FONT></STRONG> </FONT></FONT></LI> </UL> <P><FONT face="courier new,courier"><FONT face="arial,helvetica,sans-serif"><FONT color="#000000"><EM><FONT face="courier new,courier"><FONT face="arial,helvetica,sans-serif">Important: <FONT face="courier new,courier"><STRONG>ACTIVITY_NAME</STRONG> </FONT>must match the activity name in the ADF notebook activity, otherwise a query in Log Analytics will not be able to match up the logs between the two systems. </FONT></FONT></EM></FONT></FONT></FONT></P> <H2>Databricks Notebook Widgets</H2> <P>Your Databricks notebooks will look something like this:</P> <P> </P> <LI-CODE lang="python">from databricks_loganalytics.log_analytics import notebook_logger dbutils.widgets.text("PIPELINE_RUN_ID","","PIPELINE_RUN_ID") dbutils.widgets.text("PIPELINE_NAME","","PIPELINE_NAME") dbutils.widgets.text("ACTIVITY_NAME","","ACTIVITY_NAME") PIPELINE_RUN_ID=dbutils.widgets.get("PIPELINE_RUN_ID") PIPELINE_NAME=dbutils.widgets.get("PIPELINE_NAME") ACTIVITY_NAME=dbutils.widgets.get("ACTIVITY_NAME") logger = notebook_logger(PIPELINE_RUN_ID, PIPELINE_NAME, ACTIVITY_NAME) # <some notebook code> logger.log_info("some log message") # <more notebook code> # last cell logger.log_notebook_output(Out)</LI-CODE> <P> </P> <P><EM>Important note: <STRONG><FONT face="courier new,courier">Out</FONT></STRONG> is a variable that Databricks Notebooks populates automatically, you will not need to define it and make sure not to overwrite it!</EM></P> <H2>Example Log Analytics Query</H2> <P>Here is a Kusto query you can run in Log Analytics to get all the logs for a given pipeline run and child pipeline activities. Make sure to fill out the <FONT face="courier new,courier"><STRONG>ResourceId</STRONG></FONT> and a pipeline <STRONG><FONT face="courier new,courier">RunId</FONT></STRONG><FONT face="arial,helvetica,sans-serif"> (e.g. <FONT face="courier new,courier">92d3f580-3cba-4feb-a1c1-c65bddd83b2f</FONT>) </FONT>with your own. Also we used <STRONG><FONT face="courier new,courier">dbrx_notebook_logs</FONT></STRONG>, but make sure to change it if you set the <FONT face="courier new,courier"><STRONG>LOG_TABLE_NAME</STRONG></FONT> env variable to something else.</P> <P> </P> <LI-CODE lang="javascript">// -------------------------------------------------------- // Enter the run id below to get the related and child runs // -------------------------------------------------------- let runid = "<a pipeline runid>"; // get the child pipeline run IDs let relatedRunIds = AzureDiagnostics | where ResourceId contains "<your ResourceId>" and Category == "PipelineRuns" | mv-expand pred = todynamic(Predecessors_s) | project rootRunId = coalesce(pred.PipelineRunId, runId_g), runId_g | where rootRunId == runid | union (print runId_g = runid) | distinct runId_g; // Run the main query AzureDiagnostics | where ResourceId contains "<your ResourceId>" and pipelineRunId_g in (relatedRunIds) and (status_s == "Succeeded" or status_s == "Failed") | join kind=leftouter dbrx_notebook_logs on $left.pipelineRunId_g == $right.pipelineRunId_g_g and $left.pipelineName_s == $right.pipeline_s and $left.activityName_s == $right.activity_s | project TimeGenerated,pipelineRunId = coalesce(pipelineRunId_g, groupId_g), pipelineName_s, activityName_s, status_s, Message = coalesce(Message1, Message) | where Message != "Accepted" | order by TimeGenerated asc</LI-CODE> <P> </P> <P>The above query will:</P> <OL> <LI>Find all the child pipeline run IDs associated with the pipeline run ID you passed.</LI> <LI>Find all the ADF logs associated with those run IDs.</LI> <LI>Join them with the Databricks notebook logs.</LI> <LI>Massage them for a clean and ordered output.</LI> </OL> <H1>Conclusion</H1> <P>Congrats! You've set up logging! This solution is more of a proof-of-concept and the code provided will not be maintained long-term. Feel free to fork the code provided or copy it over to your codebase and add methods that achieve your own ends. I rolled everything into a .whl file for convenience, but plan to get your hands dirty! According to the needs of your organization, expect to customize this custom logging code and make it your own. </P> <P><BR /><LI-TOC indent="15" liststyle="disc" maxheadinglevel="2"></LI-TOC></P> Mon, 13 May 2024 17:26:00 GMT Dan_Z 2024-05-13T17:26:00Z Unified Logging for Databricks Notebooks and ADF with Azure Log Analytics https://community.databricks.com/t5/technical-blog/unified-logging-for-databricks-notebooks-and-adf-with-azure-log/ba-p/64717 <P>How to get all your Azure Data Factory (ADF) and Databricks Notebooks logs in the same place for once!</P> Mon, 13 May 2024 17:26:00 GMT https://community.databricks.com/t5/technical-blog/unified-logging-for-databricks-notebooks-and-adf-with-azure-log/ba-p/64717 Dan_Z 2024-05-13T17:26:00Z