Does Spline support Kudu and Iceberg tables?

[LiamWong2024]

In our test, when Spark 3.1 access data stored in Hive, Spline works well. But for Kudu or Iceberg tables, the ExecutionPlan and ExecutionEvent are both empty. Could you please give me some advice on how to capture lineage from queries against Kudu and Iceberg tables?

[wajda]

It might require a custom Spline plugin.
Enable DEBUG log level and search for the text "was not recognized as a write-command" in the logs.
If there is one then a plugin needs to be developed.
The TRACE logging will also dump the text representation of the Spark internal execution plan structure. Look for the line "**** OBJECT DUMP BEGIN ****".

[LiamWong2024]

Thanks for your answer.
Here is my configuration in spline.default.properties.
spline.lineageDispatcher=console

And here are the parameters when I submit the Spark job.
--jars hdfs://xxx/spark-3.1-spline-agent-bundle_2.12-0.7.2.jar
--conf spark.sql.queryExecutionListeners=za.co.absa.spline.harvester.listener.SplineQueryExecutionListener

But there is an exception info in Spark Driver stderr.log as following. I guess this is the reason that results in the empty ExecutionPlan and ExecutionEvent. Could you tell me how to address it?

Log Type: stderr

Log Upload Time: Wed Jan 19 14:49:56 +0800 2022

Log Length: 4345

OpenJDK 64-Bit Server VM warning: ignoring option MaxPermSize=256m; support was removed in 8.0
log4j:WARN Continuable parsing error 29 and column 23
log4j:WARN The content of element type "log4j:configuration" must match "(renderer*,throwableRenderer?,appender*,plugin*,(category|logger)*,root?,(categoryFactory|loggerFactory)?)".
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
Exception in thread "spark-listener-group-shared" java.lang.InternalError: Malformed class name
	at java.lang.Class.getSimpleName(Class.java:1330)
	at za.co.absa.spline.harvester.logging.ObjectStructureDumper$.isReadyForPrint(ObjectStructureDumper.scala:122)
	at za.co.absa.spline.harvester.logging.ObjectStructureDumper$.objectToStringRec(ObjectStructureDumper.scala:81)
	at za.co.absa.spline.harvester.logging.ObjectStructureDumper$.dump(ObjectStructureDumper.scala:39)
	at za.co.absa.spline.harvester.logging.ObjectStructureLogging.$anonfun$logObjectStructure$1(ObjectStructureLogging.scala:41)
	at scala.util.Try$.apply(Try.scala:213)
	at za.co.absa.spline.harvester.logging.ObjectStructureLogging.logObjectStructure(ObjectStructureLogging.scala:41)
	at za.co.absa.spline.harvester.logging.ObjectStructureLogging.logWarningObjectStructure(ObjectStructureLogging.scala:31)
	at za.co.absa.spline.harvester.logging.ObjectStructureLogging.logWarningObjectStructure$(ObjectStructureLogging.scala:30)
	at za.co.absa.spline.harvester.LineageHarvester.logWarningObjectStructure(LineageHarvester.scala:46)
	at za.co.absa.spline.harvester.LineageHarvester.tryExtractWriteCommand(LineageHarvester.scala:158)
	at za.co.absa.spline.harvester.LineageHarvester.harvest(LineageHarvester.scala:64)
	at za.co.absa.spline.harvester.QueryExecutionEventHandler.onSuccess(QueryExecutionEventHandler.scala:42)
	at za.co.absa.spline.harvester.listener.SplineQueryExecutionListener.$anonfun$onSuccess$2(SplineQueryExecutionListener.scala:40)
	at za.co.absa.spline.harvester.listener.SplineQueryExecutionListener.$anonfun$onSuccess$2$adapted(SplineQueryExecutionListener.scala:40)
	at scala.Option.foreach(Option.scala:407)
	at za.co.absa.spline.harvester.listener.SplineQueryExecutionListener.$anonfun$onSuccess$1(SplineQueryExecutionListener.scala:40)
	at za.co.absa.spline.harvester.listener.SplineQueryExecutionListener.withErrorHandling(SplineQueryExecutionListener.scala:49)
	at za.co.absa.spline.harvester.listener.SplineQueryExecutionListener.onSuccess(SplineQueryExecutionListener.scala:40)
	at org.apache.spark.sql.util.ExecutionListenerBus.doPostEvent(QueryExecutionListener.scala:155)
	at org.apache.spark.sql.util.ExecutionListenerBus.doPostEvent(QueryExecutionListener.scala:131)
	at org.apache.spark.util.ListenerBus.postToAll(ListenerBus.scala:117)
	at org.apache.spark.util.ListenerBus.postToAll$(ListenerBus.scala:101)
	at org.apache.spark.sql.util.ExecutionListenerBus.postToAll(QueryExecutionListener.scala:131)
	at org.apache.spark.sql.util.ExecutionListenerBus.onOtherEvent(QueryExecutionListener.scala:135)
	at org.apache.spark.scheduler.SparkListenerBus.doPostEvent(SparkListenerBus.scala:100)
	at org.apache.spark.scheduler.SparkListenerBus.doPostEvent$(SparkListenerBus.scala:28)
	at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
	at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
	at org.apache.spark.util.ListenerBus.postToAll(ListenerBus.scala:117)
	at org.apache.spark.util.ListenerBus.postToAll$(ListenerBus.scala:101)
	at org.apache.spark.scheduler.AsyncEventQueue.super$postToAll(AsyncEventQueue.scala:105)
	at org.apache.spark.scheduler.AsyncEventQueue.$anonfun$dispatch$1(AsyncEventQueue.scala:105)
	at scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23)
	at scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
	at org.apache.spark.scheduler.AsyncEventQueue.org$apache$spark$scheduler$AsyncEventQueue$$dispatch(AsyncEventQueue.scala:100)
	at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.$anonfun$run$1(AsyncEventQueue.scala:96)
	at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1381)
	at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.run(AsyncEventQueue.scala:96)

[wajda]

Seems like it's Java bug that was fixed in Java 9+
Can you try it with Java 11 for example?

[LiamWong2024]

Thanks a lot, I will have a try.

[julianzlzhang]

It looks working well when spark runs on iceberg tables. But when MERGE INFO is used, no lineage is captured.
Does the project have the plan to support it? Or where should I add code if I do it by myself?
BTW, is there any a document descripted how to debug the code when running it on spark cluster?

[wajda]

Hopefully this can help - #394