AbsaOSS / AbsaOSS/spline-spark-agent

Unity Catalog Support to get the Notebook details

Open
#780 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Scala
Stars
210
Forks
102
Avg merge
1d 1h
Merged PRs (30d)
1

Description

Hi
i creaed a code on the notebook to grab the Notebook data into the execution plan

```scala
import scala.util.parsing.json.JSON
import za.co.absa.spline.harvester.SparkLineageInitializer._
import za.co.absa.spline.agent.AgentConfig
import za.co.absa.spline.harvester.postprocessing.AbstractPostProcessingFilter
import za.co.absa.spline.harvester.postprocessing.PostProcessingFilter
import org.apache.commons.configuration.Configuration
import za.co.absa.spline.harvester.conf.StandardSplineConfigurationStack
import za.co.absa.spline.harvester.HarvestingContext
import za.co.absa.spline.producer.model.ExecutionPlan
import za.co.absa.spline.producer.model.ExecutionEvent
import za.co.absa.spline.producer.model.ReadOperation
import za.co.absa.spline.producer.model.WriteOperation
import za.co.absa.spline.producer.model.DataOperation
import za.co.absa.spline.harvester.ExtraMetadataImplicits._
import za.co.absa.spline.harvester.SparkLineageInitializer._

val notebookInformationJson = dbutils.notebook.getContext.toJson
val outerMap = JSON.parseFull(notebookInformationJson).getOrElse(0).asInstanceOf[Map[String,String]]
val tagMap = outerMap("tags").asInstanceOf[Map[String,String]]
val extraContextMap = outerMap("extraContext").asInstanceOf[Map[String,String]]
val notebookPath = extraContextMap("notebook_path").split("/")
//val workspaceUrl=tagMap("browserHostName")
var workspaceUrl="https://adb-7614304971745696.16.azuredatabricks.net"

val workspaceName=dbutils.notebook().getContext().notebookPath.get
val notebookURL = tagMap("browserHostName")+"/?o="+tagMap("orgId")+tagMap("browserHash")
val user = tagMap("user")
val name = notebookPath(notebookPath.size-1)
val notebookInfo = Map("notebookURL" -> notebookURL,
"user" -> user,
"workspaceName" ->workspaceName,
"workspaceUrl" -> workspaceUrl,
"name" -> name,
"mounts" -> dbutils.fs.ls("/FileStore/tables").map(_.path),
"timestamp" -> System.currentTimeMillis)
val notebookInfoJson = scala.util.parsing.json.JSONObject(notebookInfo)

class CustomFilter extends PostProcessingFilter {
def this(conf: Configuration) = this()

override def processExecutionEvent(event: ExecutionEvent, ctx: HarvestingContext): ExecutionEvent =
event.withAddedExtra(Map("foo" -> "bar"))

override def processExecutionPlan(plan: ExecutionPlan, ctx: HarvestingContext ): ExecutionPlan =
plan.withAddedExtra(Map( "notebookInfo" -> notebookInfoJson))

override def processReadOperation(op: ReadOperation, ctx: HarvestingContext ): ReadOperation =
op.withAddedExtra(Map("foo" -> "bar"))

override def processWriteOperation(op: WriteOperation, ctx: HarvestingContext): WriteOperation =
op.withAddedExtra(Map("foo" -> "bar"))

override def processDataOperation(op: DataOperation, ctx: HarvestingContext ): DataOperation =
op.withAddedExtra(Map("foo" -> "bar"))
}

val myInstance = new CustomFilter()

spark.enableLineageTracking(
AgentConfig.builder()
.postProcessingFilter(myInstance)
.build()
)
```

But i get the error
```
Proto Json module is unavailable on classpath, proto serializer not enabled.
java.util.NoSuchElementException: key not found: browserHash
java.util.NoSuchElementException: key not found: browserHash
scala.collection.MapLike.default(MapLike.scala:236)
scala.collection.MapLike.default$(MapLike.scala:235)
scala.collection.AbstractMap.default(Map.scala:65)
scala.collection.MapLike.apply(MapLike.scala:144)
scala.collection.MapLike.apply$(MapLike.scala:143)
scala.collection.AbstractMap.apply(Map.scala:65)
ammonite.$sess.cmd4$Helper.(cmd4.sc:28)
ammonite.$sess.cmd4$.(cmd4.sc:7)
ammonite.$sess.cmd4$.(cmd4.sc:-1)
```

and if i remove also browserHash
i get

```
java.lang.NoClassDefFoundError: org/apache/spark/sql/util/QueryExecutionListener
java.lang.NoClassDefFoundError: org/apache/spark/sql/util/QueryExecutionListener
za.co.absa.spline.harvester.SparkLineageInitializer$SplineSparkSessionWrapper.enableLineageTracking(SparkLineageInitializer.scala:105)
ammonite.$sess.cmd5$Helper.(cmd5.sc:64)
ammonite.$sess.cmd5$.(cmd5.sc:7)
ammonite.$sess.cmd5$.(cmd5.sc:-1)
java.lang.ClassNotFoundException: org.apache.spark.sql.util.QueryExecutionListener
java.net.URLClassLoader.findClass(URLClassLoader.java:387)
java.lang.ClassLoader.loadClass(ClassLoader.java:419)
sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
java.lang.ClassLoader.loadClass(ClassLoader.java:352)
za.co.absa.spline.harvester.SparkLineageInitializer$SplineSparkSessionWrapper.enableLineageTracking(SparkLineageInitializer.scala:105)
ammonite.$sess.cmd5$Helper.(cmd5.sc:64)
ammonite.$sess.cmd5$.(cmd5.sc:7)
ammonite.$sess.cmd5$.(cmd5.sc:-1)
```

Did someone saw this issue before (in Unity Catalog Cluster)

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.