Optionally capture memory only actions like collect(), count() etc

[ganeshnikumbh]

Hi,

I have code where I am collecting the spark DF rows into a variable and then iterating over those rows to insert in Cosmos DB graph. When I run this notebook I expect to see some lineage generated in driver logs (I tried setting the dispatcher to console as well as logging). But I do not see anything.
But I replace the collect action with writing to a file in Azure datalake, then I can see the lineage details in driver logs.
Any idea what could be the issue?

[wajda]

This is as designed - only persistent actions lineage is captured. The reason is that the data stored into memory doesn't have any stable identifier, so the lineage of it might only be useful temporarily, but not quite in a long term perspective.

On the other hand, when thinking about it after some time passed I acknowledge the existing of a valid use case for that.
I think about adding a configuration property via that the user would be able to enable capturing lineage from memory only actions.

[ganeshnikumbh]

@wajda , thanks for clarifying. I hope to see the feature soon :)..

[yuribogomolov]

+100 to the feature request.
We found multiple coverage gaps caused by this. The pattern is quite common for data quality jobs, validate-and-publish workflows and ML flows (not all ML methods are supported by Spark and data scientists usually aggregate data in Spark and use df.toPandas() to collect data and run a model before moving data back to Spark).

[wajda]

The lineage data model requires existence of a target data source that is represented by a URI.
I guess we could simply use a made up URI that would contain a host name, a type and a hash code of a result object/collection. Something like memory://1.2.3.4/[Lorg.apache.spark.sql.Row;@486e6b30

[yuribogomolov]

Yeah, that works. I guess multiple steps need changes to support "read-only" actions. collect and toPandas are implemented outside of Catalyst, and the only action-specific information provided by the Spark QueryListener is available in funcName, but it's not propagated to SplineAgent and downstream classes at this point.

[wajda]

Fixed.
Capturing of non-persistent actions are disabled by default. To enable it the following property needs to be set:
as YAML:

spline:
    plugins:
        za.co.absa.spline.harvester.plugin.embedded.NonPersistentActionsCapturePlugin:
            enabled: true

or in a properties format:

spline.plugins.za.co.absa.spline.harvester.plugin.embedded.NonPersistentActionsCapturePlugin.enabled=true