lineage from Databricks delta tables

[timteru]

Question

I'm trying to capture lineage from queries against Databricks tables (see specifics below), for example just using the console dispatcher. For a simple INSERT INTO operation, I'm encountering the following error. Is the error because I haven't configured something correctly, or is it possible that the DataSourceV2Plugin/DatabricksPlugin doesn't understand these operations on Databricks delta tables?

21/09/02 22:02:58 ERROR SplineQueryExecutionListener: Unexpected error occurred during lineage processing for application: Databricks Shell #app-20210902220020-0000
scala.MatchError: DeltaTableV2(org.apache.spark.sql.SparkSession@52662588,dbfs:/dbfs/mnt/workspaces/CMA/Reference/PreProd/Users/twhittem/versioned_data/test_lineage/ver=01,Some(CatalogTable(
Database: cma_users_twhittem
Table: test_lineage
Owner: root
Created Time: Wed Sep 01 21:45:15 UTC 2021
Last Access: UNKNOWN
Created By: Spark 3.0.1
Type: EXTERNAL
Provider: delta
Location: dbfs:/dbfs/mnt/workspaces/CMA/Reference/PreProd/Users/twhittem/versioned_data/test_lineage/ver=01
Serde Library: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
InputFormat: org.apache.hadoop.mapred.SequenceFileInputFormat
OutputFormat: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
Partition Provider: Catalog)),Some(cma_users_twhittem.test_lineage),None) (of class com.databricks.sql.transaction.tahoe.catalog.DeltaTableV2)
	at za.co.absa.spline.harvester.plugin.embedded.DataSourceV2Plugin.za$co$absa$spline$harvester$plugin$embedded$DataSourceV2Plugin$$extractSourceIdFromTable(DataSourceV2Plugin.scala:133)
	at za.co.absa.spline.harvester.plugin.embedded.DataSourceV2Plugin.za$co$absa$spline$harvester$plugin$embedded$DataSourceV2Plugin$$extractSourceIdFromRelation(DataSourceV2Plugin.scala:127)
	at za.co.absa.spline.harvester.plugin.embedded.DataSourceV2Plugin$$anonfun$2.applyOrElse(DataSourceV2Plugin.scala:68)
	at za.co.absa.spline.harvester.plugin.embedded.DataSourceV2Plugin$$anonfun$2.applyOrElse(DataSourceV2Plugin.scala:54)
	at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175)
	at scala.PartialFunction$Lifted.apply(PartialFunction.scala:228)
	at scala.PartialFunction$Lifted.apply(PartialFunction.scala:224)
	at za.co.absa.spline.harvester.builder.write.PluggableWriteCommandExtractor.asWriteCommand(PluggableWriteCommandExtractor.scala:44)
	at za.co.absa.spline.harvester.LineageHarvester.$anonfun$harvest$4(LineageHarvester.scala:64)
	at scala.util.Try$.apply(Try.scala:213)
	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:86)
	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:1523)
	at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.run(AsyncEventQueue.scala:96)

Configuration

• Databricks 7.3 runtime (Spark 3.0.1, Scala 2.12)
• spark-3.0-spline-agent-bundle_2.12-1.0.0-SNAPSHOT.jar
• spark.sql.queryExecutionListeners=za.co.absa.spline.harvester.listener.SplineQueryExecutionListener
• spark.spline.lineageDispatcher=console

Set up and test query

create table if not exists cma_users_twhittem.test_lineage (AccountID string) using DELTA location "/dbfs/mnt/workspaces/CMA/Reference/PreProd/Users/twhittem/versioned_data/test_lineage/ver=01"

insert into cma_users_twhittem.test_lineage values ("test_id") 

Follow-up

I can get this to work with a very small modification to the DataSourceV2Plugin:

private def extractSourceIdFromTable(table: AnyRef): SourceIdentifier = table match {

   ...

   case `_: DatabricksDeltaTableV2Command`(dbdt) =>
      val uri = extractFieldValue[Option[String]](dbdt, "tableIdentifier")
        .getOrElse(extractFieldValue[String](dbdt, "path"))
      SourceIdentifier(Some("databricksDelta"), uri)
  }
}

object DataSourceV2Plugin {

  ...

  object `_: DatabricksDeltaTableV2Command` extends SafeTypeMatchingExtractor[AnyRef](
    "com.databricks.sql.transaction.tahoe.catalog.DeltaTableV2"
  )
}

I could not accomplish this with a modification to the existing DatabricksPlugin, I'm guessing because com.databricks.sql.transaction.tahoe.catalog.DeltaTableV2 must implement org.apache.spark.sql.catalyst.plans.logical.V2WriteCommand and is therefore captured by the DataSourceV2Plugin?

[cerveada]

Your guess is correct. Right now, all V2 data sources are handled in the DataSourceV2Plugin, no matter the type.

This was fixed in PR #291. It's not released yet.

Could you try whether build fixes the issue for you? The bundles are available here:
https://teamcity.jetbrains.com/buildConfiguration/OpenSourceProjects_AbsaOSS_SplineAgentSpark_PullRequestsSpark30scala212/3599210?buildTab=artifacts#%2Fbundle-3.0%2Ftarget

[timteru]

Thank you! Confirming that the changes in PR #291 fix the issue with simple inserts. I am currently testing more complicated cases.