Store and print Lineage information in variable in code page

[tariniroutey]

HI Team,

Just wanted to know if i can write the Lineage information to any variable in the code page it self if possible. Just explain in detail

System.setProperty("spline.mode", "REQUIRED")
System.setProperty("spark.sql.queryExecutionListeners", "za.co.absa.spline.harvester.listener.SplineQueryExecutionListener")
_**// System.setProperty("spline.producer.url","https://**********/api/func/producer")  -- Commenting it as i don't want to send to producer.**_

import za.co.absa.spline.harvester.SparkLineageInitializer._
spark.enableLineageTracking()

//Sample code to generate lineage
case class emp(emp_id: Int, emp_name: String,emp_sal: Int, addrId: Int)
case class Address(id: Int, address: String)
Seq(emp(1, "A",100,1),emp(2, "B",200,2),emp(3, "C",300,3)).toDS.write.mode("overwrite").parquet("/emp")
Seq(Address(1, "A1"),Address(2, "A2"),Address(3, "A3")).toDS.write.mode("overwrite").parquet("/address")
val employee = spark.read.parquet("/emp")
val addresses = spark.read.parquet("/address")
employee .join(addresses) .where(addresses("id") === employee("addrId")).select("emp_id","emp_name","emp_sal", "address")
  .write.mode("overwrite").parquet("/emp_address")

I want to store the complete lineage json in variable and do further cleaning after my code execution.
Any sample code reference would be great help.

Thank You.

[cerveada]

I want to store the complete lineage json in variable

There are several dispatchers available, check the docs:
https://github.com/AbsaOSS/spline-spark-agent#lineage-dispatchers

If none of them meets your needs, you can create your own.

and do further cleaning after my code execution.

Post processing filter may be useful for this. It is applied before the dispatcher sends/stores the lineage data.
https://github.com/AbsaOSS/spline-spark-agent#post-processing-filters

[wajda]

I want to store the complete lineage json in variable

The short answer is yes, as Adam said, but there are some but's. Let me elaborate on it. Spline agent is called by Spark driver from a separate thread, so to pull the lineage content into a variable in your notebook page or a shell you need to do it in concurrent manner. With Spark 2.x it's easier because the actions were blocking and by the time the control is returned the Spline work has already done, and all the dispatchers are called, so you can expect the lineage to be captured. However in Spark 3+ the event listeners are processed asynchronously to the actions, hence you need to implement some sort of synchronization and wait until the lineage content is ready and written into your variable. This is not that straight-forward, but is doable. We do it in our integration-tests. Take a look at the LineageCaptor class, and the usage in some tests, e.g. BasicIntegrationTests