Errors while using UserExtraMeta or CustomPsotProcessingFilter

[rohank-dev]

Hello All,

Good Day!

I am trying to add custom parameters to the lineage by using 2 approaches mentioned below -

1. Using MetadataCollectingFilter
2. Using Custom Post Processing Filters
   But I am getting errors while using them.

I have created a custom library with the additional details to process Filters. In both the approaches following steps are common

"conf" : {
   "spark.jars.packages": "org:splinesparkplugin_2.12:0.1"
}

spark.enableLineageTracking(
    AgentConfig.builder()
    .splineMode(SplineMode.ENABLED)
    .config("spline.postProcessingFilter","<filterName>")
  //Other filter related config as mentioned below
   )

Approach 1: (Using MetadataCollectingFilter)

1. Use spline.postProcessingFilter=userExtraMeta
2. for spline.postProcessingFilter.userExtraMeta.rules I tried 2 variations (none worked) -
   a. defining a String with rule and passing it to above key as a parameter -
   eg:

    val ruleString =
      """
        |{
        |    "executionPlan": {
        |        "labels": {
        |            "qux": 42,
        |            "tags": [ "aaa", "bbb", null, "ccc" ],
        |            "foo1": null,
        |            "foo2": [],
        |            "foo3": [null],
        |            "notFlat": ["a", ["b", "c"]]
        |        }
        |    },
        |    "executionEvent": {
        |        "labels": {
        |            "qux": 42
        |        }
        |    }
        |}
        |""".stripMargin

b.

val ruleString = "{\"executionPlan\":{\"foo\":{\"qux\":42,\"tags\":[\"aaa\",\"bbb\",\"ccc\"]}}"

In both cases I am getting below error -

ERROR SparkLineageInitializer: Spline initialization failed! Spark Lineage tracking is DISABLED.
za.co.absa.spline.shaded.org.json4s.package$MappingException: Expected object but got JNothing
	at za.co.absa.spline.shaded.org.json4s.reflect.package$.fail(package.scala:95)
	at za.co.absa.spline.shaded.org.json4s.Extraction$.extract(Extraction.scala:379)
	at za.co.absa.spline.shaded.org.json4s.Extraction$.extract(Extraction.scala:39)
	at za.co.absa.spline.shaded.org.json4s.ExtractableJsonAstNode.extract(ExtractableJsonAstNode.scala:21)
	at za.co.absa.spline.harvester.json.HarvesterJsonSerDe$impl$JsonToEntity.fromJson(HarvesterJsonSerDe.scala:36)
	at za.co.absa.spline.harvester.postprocessing.metadata.MetadataCollectingFilter$.za$co$absa$spline$harvester$postprocessing$metadata$MetadataCollectingFilter$$createRuleDefs(MetadataCollectingFilter.scala:86)

Approach 2: (Using Custom Post Processing Filters)

1. Created a CustomPostProcessingFilter (conf: Configuration) extends PostProcessingFilter
2. Overrided the methods where extra elements needed to be added
3. Set below values -

spline.postProcessingFilter=customPostProcessingFilter 
spline.postProcessingFilter.customPostProcessingFilter.className=my.awesome.customPostProcessingFilter

But even after the class file file existed in the supplied jar, I got below exception -

java.lang.ClassNotFoundException: my.awesome.customPostProcessingFilter
	at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
	at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
	at java.lang.Class.forName0(Native Method)
	at java.lang.Class.forName(Class.java:264)
	at za.co.absa.commons.HierarchicalObjectFactory.instantiate(HierarchicalObjectFactory.scala:41)

Please suggest

[cerveada]

What Spark version are you on?

[rohank-dev]

The spark version used is 3.2

[wajda]

For no.1 (userExtraMeta.rules) - escape every comma (as \,). It's not clearly stated in the doc, but when passing configuration via Java properties the comma is used as a separator under the hood and must be explicitly escaped. I'll file an issue for that for the future.

For no.2 (ClassNotFoundException) - my.awesome.customPostProcessingFilter is supposed to be your awesome filter, that you have implemented :) Did you do that? Make sure it's indeed in the Spark driver class path.

[rohank-dev]

For no.1 (userExtraMeta.rules) - escape every comma (as \,).

As mentioned my config is now as below -

 .config("spline.postProcessingFilter.userExtraMeta.rules", "{\"executionPlan\":{\"foo\":{\"qux\":42\\,\"tags\":[\"aaa\"\\,\"bbb\"\\,\"ccc\"]}}" )

It didnt gave any error and was processed, however I was not able to find the extra elements -

{"planId":"<Some_ID>","timestamp":1686149827400,"durationNs":338569816,"extra":{"appId":"application_1_0222","user":"rohan","readMetrics":{},"writeMetrics":{}}}

For no.2 (ClassNotFoundException) - my.awesome.customPostProcessingFilter is supposed to be your awesome filter, that you have implemented :) Did you do that? Make sure it's indeed in the Spark driver class path.

Yes it is present in the supplied plugin library and I created customPostProcessingFilter in the mentioned package by extending PostProcessingFilter as per the documentaion

[wajda]

1. Use { "executionPlan": { "extra": { "foo": { ....
2. ClassNotFoundException can only mean one thing - the class is not in the class path. Please check your runtime setup one more time.

P.S. Please use backticks or triple back ticks when posting code snippets, configs, logs etc. Otherwise your message gets almost unreadable when posted on Github. Also using "Preview" button before sending a message is a good practice.

[cerveada]

https://docs.github.com/en/get-started/writing-on-github/working-with-advanced-formatting/creating-and-highlighting-code-blocks

[rohank-dev]

Thanks a lot for your comment.

Use { "executionPlan": { "extra": { "foo": { ....

I have made sure that for all the below mentioned scenario I have set the filter as -

.config("spline.postProcessingFilter", userExtraMeta)

I updates the code as below -

.config("spline.postProcessingFilter.userExtraMeta.rules", "{\"executionPlan\": { \"extra\": { \"appName\": \"session.sparkContext.appName\"\\,  \"attemptId\": \"session.sparkContext.applicationAttemptId\"\\,  \"tag\": \"spline-spark-plugin-0.1.0.1\"  }  }" )

The elements are getting added now. Could you please suggest how I can get the appName and attemptId from the session object, since in the above scenario the output will be replaced as a String and be represented as follows -

extra":{"readMetrics":{},"tag":"spline-spark-plugin-0.1.0.1","appName":"session.sparkContext.appName","attemptId":"session.sparkContext.applicationAttemptId","appId":"application_1","user":"rohan","writeMetrics":{}}}

[cerveada]

Without the slashes:

"appName":{ "$js":"session.sparkContext.appName()" }

• The $js will tell the Agent that the following key is not just a string but code to avaluate as javascript in this case.
• there are some variable abvailable by default session is org.apache.spark.sql.SparkSession
• from the session you get sparkContext.appName()
• appName is a function not a val so you need appName() to call it. Maybe it would work without it I am not sure.

There is some documentation here describing some of this stuff, though it's not complete so feel free to ask if there is something missing.

[cerveada]

Just tried it by myself and what worked for me was:

"appName": { "$js":"session.sparkContext().appName()" }

[rohank-dev]

Thank you so much it worked and you were correct we need to use sparkContext as well as appName as a function. This worked for appName as well as AttemptId.
A quick question though while writing in spline.yaml if I wish to pass the rules directly as the value rather than the file path, How can I define the above config.

userExtraMeta:
      className: za.co.absa.spline.harvester.postprocessing.metadata.MetadataCollectingFilter
      # rules are defined in a JSON format
      # can be a json string or url to json file
      rules: >
        {
          "executionPlan": {
            "extra": {
                  "user": "session.sparkContext().sparkUser()",
                  "tag": "spline-spark-plugin-0.1.0.1"
            }
          },
          "executionEvent": {
            "extra": {
                  "appName": "session.sparkContext().appName()",
                  "attemptId": "session.sparkContext().applicationAttemptId()",
                  "tag": "spline-spark-plugin-0.1.0.1"
            }
          },
          "operation": {
            "extra": {
                  "user": "session.sparkContext().sparkUser()",
                  "tag": "spline-spark-plugin-0.1.0.1"
            }
          },
          "read": {
            "extra": {
                  "user": "session.sparkContext().sparkUser()",
                  "tag": "spline-spark-plugin-0.1.0.1"
            }
          },
          "write": {
            "extra": {
                  "user": "session.sparkContext().sparkUser()",
                  "tag": "spline-spark-plugin-0.1.0.1"
            }
          }
        }

[cerveada]

It doesn't matter if you specify the json directly or in a file, the syntax is the same. Only in some places like .config(...) escaping is needed.

Looking at your last json, you can't just do "code" an expect it to execute. You must do { "$js":"code" }.

For example:

"session.sparkContext().sparkUser()" => { "$js":"session.sparkContext().sparkUser()" }.

See this example of yaml config with several rules: https://github.com/AbsaOSS/spline-spark-agent/blob/develop/examples/src/main/resources/spline.yaml

[rohank-dev]

Agreed and thanks a lot for your answer

[cerveada]

You are welcome!

I improved the documentation a little bit to add more info about the filter, especially how to do conditional filtering.
https://github.com/AbsaOSS/spline-spark-agent#using-metadatacollectingfilter