UserExtraAppendingPostProcessingFilter #228

ronlesteve · 2021-06-02T18:20:29Z

ronlesteve
Jun 2, 2021

Hello, I have been trying to run the following code with the new spline jar: za.co.absa.spline.agent.spark:spark-3.0-spline-agent-bundle_2.12:0.6.0 but have been getting errors specific to UserExtraMetadataProvider which has been deprecated in the newer versions. I have also tried replacing UserExtraMetadataProvider with UserExtraAppendingPostProcessingFilter using the code shown below this first code block but still getting the errors. Can you please validate and share how to properly write the post processing filter code using the new spline bundle.

import za.co.absa.spline.harvester.conf.StandardSplineConfigurationStack
import za.co.absa.spline.harvester.extra.UserExtraMetadataProvider
import za.co.absa.spline.harvester.HarvestingContext
import org.apache.commons.configuration.Configuration
import za.co.absa.spline.harvester.SparkLineageInitializer._
import za.co.absa.spline.harvester.conf.DefaultSplineConfigurer
import za.co.absa.spline.producer.model._
import scala.util.parsing.json.JSON

val splineConf: Configuration = StandardSplineConfigurationStack(spark)

spark.enableLineageTracking(new DefaultSplineConfigurer(splineConf) {
  //override protected def userExtraMetadataProvider = new UserExtraMetaDataProvider {
  //val test = dbutils.notebook.getContext.notebookPath
  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 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, 
                "name" -> name,
                "mounts" -> dbutils.fs.ls("/mnt").map(_.path),
                "timestamp" -> System.currentTimeMillis)
  val notebookInfoJson = scala.util.parsing.json.JSONObject(notebookInfo)
  
  override protected def userExtraMetadataProvider: UserExtraMetadataProvider = new UserExtraMetadataProvider {
    override def forExecEvent(event: ExecutionEvent, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar1")
    override def forExecPlan(plan: ExecutionPlan, ctx: HarvestingContext): Map[String, Any] = Map("notebookInfo" -> notebookInfoJson) // tilføj mount info til searchAndReplace  denne funktion indeholder infoen
    override def forOperation(op: ReadOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar3")
    override def forOperation(op: WriteOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar4")
    override def forOperation(op: DataOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar5")
  }
})

Here is the updated code that is still having errors

import za.co.absa.spline.harvester.conf.StandardSplineConfigurationStack
import za.co.absa.spline.harvester.extra.UserExtraMetadataProvider
import za.co.absa.spline.harvester.HarvestingContext
import org.apache.commons.configuration.Configuration
import za.co.absa.spline.harvester.SparkLineageInitializer._
import za.co.absa.spline.harvester.conf.DefaultSplineConfigurer
import za.co.absa.spline.producer.model._
import play.api.libs.json._

val splineConf: Configuration = StandardSplineConfigurationStack(spark)

spark.enableLineageTracking(new DefaultSplineConfigurer(splineConf) {
  val notebookInformationJson = Json.toJson(dbutils.notebook.getContext)
  val outerMap = Json.toJson(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 notebookURL = tagMap("browserHostName")+"/?o="+tagMap("orgId")+tagMap("browserHash")
  val user = tagMap("user")
  val name = notebookPath(notebookPath.size-1)
  
  
   val notebookInfo = Map("notebookURL" -> Json.toJson(notebookURL),  
                "user" -> Json.toJson(user), 
                "name" -> Json.toJson(name),
                "mounts" -> Json.toJson(dbutils.fs.ls("/mnt").map(_.path)),
                "timestamp" -> Json.toJson(System.currentTimeMillis))

val notebookInfoJson =  Json.toJson(notebookInfo)

def userExtraMetadataProvider: UserExtraAppendingPostProcessingFilter
 = new UserExtraAppendingPostProcessingFilter
  
  
  {
    def processExecutionEvent(event: ExecutionEvent, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar1")
    def processExecutionPlan (plan: ExecutionPlan, ctx: HarvestingContext): Map[String, Any] = Map("notebookInfo" -> notebookInfoJson)
    def processReadOperation(op: ReadOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar3")
    def processWriteOperation(op: WriteOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar4")
    def processDataOperation(op: DataOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar5")
  }
})

Here is the error

command-2044409137370707:12: error: not enough arguments for constructor DefaultSplineConfigurer: (sparkSession: org.apache.spark.sql.SparkSession, userConfiguration: org.apache.commons.configuration.Configuration)za.co.absa.spline.harvester.conf.DefaultSplineConfigurer.
Unspecified value parameter userConfiguration.
spark.enableLineageTracking(new DefaultSplineConfigurer(splineConf) {
                                ^
command-2044409137370707:32: error: not found: type UserExtraAppendingPostProcessingFilter
 def userExtraMetadataProvider: UserExtraAppendingPostProcessingFilter
                                ^
command-2044409137370707:33: error: not found: type UserExtraAppendingPostProcessingFilter
 = new UserExtraAppendingPostProcessingFilter
       ^
command-2044409137370707:37: error: not found: type ExecutionEvent
    def processExecutionEvent(event: ExecutionEvent, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar1")
                                     ^
command-2044409137370707:38: error: not found: type ExecutionPlan
    def processExecutionPlan (plan: ExecutionPlan, ctx: HarvestingContext): Map[String, Any] = Map("notebookInfo" -> notebookInfoJson)
                                    ^
command-2044409137370707:39: error: not found: type ReadOperation
    def processReadOperation(op: ReadOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar3")
                                 ^
command-2044409137370707:40: error: not found: type WriteOperation
    def processWriteOperation(op: WriteOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar4")
                                  ^
command-2044409137370707:41: error: not found: type DataOperation
    def processDataOperation(op: DataOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar5")
                                 ^
command-2044409137370707:36: warning: a pure expression does nothing in statement position; multiline expressions may require enclosing parentheses
  {
  ^

wajda · 2021-06-03T08:49:14Z

wajda
Jun 3, 2021
Maintainer

Your code doesn't compile for a few reasons:

You miss some imports (the error log makes it clear):

import za.co.absa.spline.producer.model.v1_1._
import za.co.absa.spline.harvester.extra.UserExtraAppendingPostProcessingFilter

The correct signature for the extra metadata provider is the following:

protected def maybeUserExtraMetadataProvider: Option[UserExtraMetadataProvider]

UserExtraAppendingPostProcessingFilter is just an adapter for the deprecated UserExtraMetadataProvider. So you still need to create an instance:
```
new UserExtraAppendingPostProcessingFilter(new UserExtraMetadataProvider() { 
  // ???
})
```

Please note that we are working on a declarative solution for capturing the extra metadata, so that most of the rules and values could be defined in the configuration, and little to no coding would be required for that. See #169

For now just use UserExtraMetadataProvider

3 replies

ronlesteve Jun 4, 2021
Author

@wajda Thanks. I have made the following modifications to the code as follows and getting the following errors. Do you have any recommendation on how to resolve the errors:

import za.co.absa.spline.harvester.conf.StandardSplineConfigurationStack
import za.co.absa.spline.harvester.extra.UserExtraAppendingPostProcessingFilter
import za.co.absa.spline.harvester.HarvestingContext
import org.apache.commons.configuration.Configuration
import za.co.absa.spline.harvester.SparkLineageInitializer._
import za.co.absa.spline.harvester.conf.DefaultSplineConfigurer
import za.co.absa.spline.producer.model._
import play.api.libs.json._
import za.co.absa.spline.producer.model.v1_1._
import za.co.absa.spline.harvester.postprocessing.PostProcessingFilter


val splineConf: Configuration = StandardSplineConfigurationStack(spark)

spark.enableLineageTracking(new DefaultSplineConfigurer(spark,splineConf) {
  val notebookInformationJson = Json.toJson(dbutils.notebook.getContext)
  val outerMap = Json.toJson(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 notebookURL = tagMap("browserHostName")+"/?o="+tagMap("orgId")+tagMap("browserHash")
  val user = tagMap("user")
  val name = notebookPath(notebookPath.size-1)
  
  
   val notebookInfo = Map("notebookURL" -> Json.toJson(notebookURL),  
                "user" -> Json.toJson(user), 
                "name" -> Json.toJson(name),
                "mounts" -> Json.toJson(dbutils.fs.ls("/mnt").map(_.path)),
                "timestamp" -> Json.toJson(System.currentTimeMillis))

val notebookInfoJson =  Json.toJson(notebookInfo)

 override protected def maybeUserExtraMetadataProvider: Option[UserExtraMetadataProvider] = new UserExtraAppendingPostProcessingFilter(new UserExtraMetadataProvider() {
    override def forExecEvent(event: ExecutionEvent, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar1")
    override def forExecPlan(plan: ExecutionPlan, ctx: HarvestingContext): Map[String, Any] = Map("notebookInfo" -> notebookInfoJson) 
    override def forOperation(op: ReadOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar3")
    override def forOperation(op: WriteOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar4")
    override def forOperation(op: DataOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar5")
  })
})

command-3309490562461377:36: error: not found: type UserExtraMetadataProvider
 override protected def maybeUserExtraMetadataProvider: Option[UserExtraMetadataProvider] = new UserExtraAppendingPostProcessingFilter(new UserExtraMetadataProvider() {
                                                               ^
command-3309490562461377:36: error: not found: type UserExtraMetadataProvider
 override protected def maybeUserExtraMetadataProvider: Option[UserExtraMetadataProvider] = new UserExtraAppendingPostProcessingFilter(new UserExtraMetadataProvider() {
                                                                                                                                           ^
command-3309490562461377:36: error: type mismatch;
 found   : <error>{override def forExecEvent(event: za.co.absa.spline.producer.model.v1_1.ExecutionEvent,ctx: za.co.absa.spline.harvester.HarvestingContext): Map[String,Any]; override def forExecPlan(plan: za.co.absa.spline.producer.model.v1_1.ExecutionPlan,ctx: za.co.absa.spline.harvester.HarvestingContext): Map[String,Any]; override def forOperation(op: za.co.absa.spline.producer.model.v1_1.ReadOperation,ctx: za.co.absa.spline.harvester.HarvestingContext): Map[String,Any]; override def forOperation(op: za.co.absa.spline.producer.model.v1_1.WriteOperation,ctx: za.co.absa.spline.harvester.HarvestingContext): Map[String,Any]; override def forOperation(op: za.co.absa.spline.producer.model.v1_1.DataOperation,ctx: za.co.absa.spline.harvester.HarvestingContext): Map[String,Any]}
 required: za.co.absa.spline.harvester.extra.UserExtraMetadataProvider
 override protected def maybeUserExtraMetadataProvider: Option[UserExtraMetadataProvider] = new UserExtraAppendingPostProcessingFilter(new UserExtraMetadataProvider() {

wajda Jun 5, 2021
Maintainer

import za.co.absa.spline.harvester.conf.StandardSplineConfigurationStack
import za.co.absa.spline.harvester.extra.UserExtraAppendingPostProcessingFilter
import za.co.absa.spline.harvester.extra.UserExtraMetadataProvider
import za.co.absa.spline.harvester.HarvestingContext
import org.apache.commons.configuration.Configuration
import za.co.absa.spline.harvester.SparkLineageInitializer._
import za.co.absa.spline.harvester.conf.DefaultSplineConfigurer
import za.co.absa.spline.producer.model._
import play.api.libs.json._
import za.co.absa.spline.producer.model.v1_1._
import za.co.absa.spline.harvester.postprocessing.PostProcessingFilter


val splineConf: Configuration = StandardSplineConfigurationStack(spark)

spark.enableLineageTracking(new DefaultSplineConfigurer(spark,splineConf) {
  val notebookInformationJson = Json.toJson(dbutils.notebook.getContext)
  val outerMap = Json.toJson(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 notebookURL = tagMap("browserHostName")+"/?o="+tagMap("orgId")+tagMap("browserHash")
  val user = tagMap("user")
  val name = notebookPath(notebookPath.size-1)


  val notebookInfo = Map("notebookURL" -> Json.toJson(notebookURL),
    "user" -> Json.toJson(user),
    "name" -> Json.toJson(name),
    "mounts" -> Json.toJson(dbutils.fs.ls("/mnt").map(_.path)),
    "timestamp" -> Json.toJson(System.currentTimeMillis))

  val notebookInfoJson =  Json.toJson(notebookInfo)

  override protected def maybeUserExtraMetadataProvider: Option[UserExtraMetadataProvider] = Some(new UserExtraMetadataProvider() {
    override def forExecEvent(event: ExecutionEvent, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar1")
    override def forExecPlan(plan: ExecutionPlan, ctx: HarvestingContext): Map[String, Any] = Map("notebookInfo" -> notebookInfoJson)
    override def forOperation(op: ReadOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar3")
    override def forOperation(op: WriteOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar4")
    override def forOperation(op: DataOperation, ctx: HarvestingContext): Map[String, Any] = Map("foo" -> "bar5")
  })
})

tkkabi Nov 21, 2021

Hi @wajda , For the above mentioned code, I am getting following error. How to fix this issue?

error: No Json serializer found for type com.databricks.backend.common.rpc.CommandContext. Try to implement an implicit Writes or Format for this type.
val notebookInformationJson = Json.toJson(dbutils.notebook.getContext)

Note -
Databricks Cluster - 7.3 LTS (includes Apache Spark 3.0.1, Scala 2.12)

Libraries installed -
za.co.absa.spline.agent.spark:spark-3.0-spline-agent-bundle_2.12:0.7.1(exclusion - org.json4s:json4s-ext_2.11:$(json4s.version)
com.typesafe.play:play-json_2.12:2.10.0-RC5

wajda · 2021-11-22T09:47:36Z

wajda
Nov 22, 2021
Maintainer

I think this is more of a question for the Play JSON than the Spline agent (if I understand it correctly, that the issue is related to your custom code Json.toJson(dbutils.notebook.getContext))

But from what I can guess the CommandContext being a trait is probably not automatically serializable by Play Json). Try to get the exact info you are interested in from the context first, and serialized that.

Actually, you don't even need to manually serialize objects that you want to include into Spline extras. If the objects are naturally serializable (simple types, case classes, standard collections etc) you can included them to extras as is - Spline will serialize them on dispatching.

1 reply

wajda Nov 24, 2021
Maintainer

Answered this on the StackOverflow - https://stackoverflow.com/a/70087887/96766

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

UserExtraAppendingPostProcessingFilter #228

{{title}}

{{editor}}'s edit

{{editor}}'s edit

Replies: 2 comments 4 replies

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

Select a reply

UserExtraAppendingPostProcessingFilter #228

ronlesteve Jun 2, 2021

Replies: 2 comments · 4 replies

wajda Jun 3, 2021 Maintainer

ronlesteve Jun 4, 2021 Author

wajda Jun 5, 2021 Maintainer

tkkabi Nov 21, 2021

wajda Nov 22, 2021 Maintainer

wajda Nov 24, 2021 Maintainer

ronlesteve
Jun 2, 2021

Replies: 2 comments 4 replies

wajda
Jun 3, 2021
Maintainer

ronlesteve Jun 4, 2021
Author

wajda Jun 5, 2021
Maintainer

wajda
Nov 22, 2021
Maintainer

wajda Nov 24, 2021
Maintainer