AbsaOSS / AbsaOSS/spline-spark-agent

Unity Catalog Support to get the Notebook details

Abierto
#780 0 comentarios 0 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Scala
Estrellas
210
Forks
102
Merge medio
1 d 1 h
PR fusionados (30 d)
1

Descripción

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)

Guía de contribución

No hay ninguna guía de contribución indexada para este repositorio

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.