Spline for databricks support of actions that dont use persist storage

[zacayd]

Hi all
As far as I know - spline records only actions that reads or writes to a persist storage
is there a way to make it record also actions of reading/writing from/to a dataframe
i used spline 1.0.4
thanks in advance

[wajda]

If you mean actions like collect() and similar, then it's possible, but you need to explicitly enable that feature.

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

[zacayd]

that you put on the configuataion of the cluster?

[wajda]

yes

[zacayd]

i put on the cluster settings
spline.plugins.za.co.absa.spline.harvester.plugin.embedded.NonPersistentActionsCapturePlugin.enabled true

and when i run the notebook i got
java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.

[wajda]

Can you show the full stack trace?

[zacayd]

Py4JJavaError                             Traceback (most recent call last)
<command-380891150632255> in <module>
     24     .option("header","true")\
     25     .option("inferschema","true")\
---> 26     .csv("/FileStore/tables/circuits-2.csv")
     27 
     28 #df.printSchema()

/databricks/spark/python/pyspark/sql/readwriter.py in csv(self, path, schema, sep, encoding, quote, escape, comment, header, inferSchema, ignoreLeadingWhiteSpace, ignoreTrailingWhiteSpace, nullValue, nanValue, positiveInf, negativeInf, dateFormat, timestampFormat, maxColumns, maxCharsPerColumn, maxMalformedLogPerPartition, mode, columnNameOfCorruptRecord, multiLine, charToEscapeQuoteEscaping, samplingRatio, enforceSchema, emptyValue, locale, lineSep, pathGlobFilter, recursiveFileLookup)
    533             path = [path]
    534         if type(path) == list:
--> 535             return self._df(self._jreader.csv(self._spark._sc._jvm.PythonUtils.toSeq(path)))
    536         elif isinstance(path, RDD):
    537             def func(iterator):

/databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1303         answer = self.gateway_client.send_command(command)
   1304         return_value = get_return_value(
-> 1305             answer, self.gateway_client, self.target_id, self.name)
   1306 
   1307         for temp_arg in temp_args:

/databricks/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
    125     def deco(*a, **kw):
    126         try:
--> 127             return f(*a, **kw)
    128         except py4j.protocol.Py4JJavaError as e:
    129             converted = convert_exception(e.java_exception)

/databricks/spark/python/lib/py4j-0.10.9-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    326                 raise Py4JJavaError(
    327                     "An error occurred while calling {0}{1}{2}.\n".
--> 328                     format(target_id, ".", name), value)
    329             else:
    330                 raise Py4JError(

Py4JJavaError: An error occurred while calling o309.csv.
: java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.
This stopped SparkContext was created at:

org.apache.spark.SparkContext.<init>(SparkContext.scala:99)
com.databricks.backend.daemon.driver.DatabricksILoop$.$anonfun$initializeSharedDriverContext$1(DatabricksILoop.scala:347)
com.databricks.backend.daemon.driver.ClassLoaders$.withContextClassLoader(ClassLoaders.scala:29)
com.databricks.backend.daemon.driver.DatabricksILoop$.initializeSharedDriverContext(DatabricksILoop.scala:347)
com.databricks.backend.daemon.driver.DatabricksILoop$.getOrCreateSharedDriverContext(DatabricksILoop.scala:277)
com.databricks.backend.daemon.driver.DriverCorral.com$databricks$backend$daemon$driver$DriverCorral$$driverContext(DriverCorral.scala:180)
com.databricks.backend.daemon.driver.DriverCorral.<init>(DriverCorral.scala:217)
com.databricks.backend.daemon.driver.DriverDaemon.<init>(DriverDaemon.scala:39)
com.databricks.backend.daemon.driver.DriverDaemon$.create(DriverDaemon.scala:211)
com.databricks.backend.daemon.driver.DriverDaemon$.wrappedMain(DriverDaemon.scala:216)
com.databricks.DatabricksMain.$anonfun$main$1(DatabricksMain.scala:106)
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
com.databricks.DatabricksMain.$anonfun$withStartupProfilingData$1(DatabricksMain.scala:336)
com.databricks.logging.UsageLogging.$anonfun$recordOperation$4(UsageLogging.scala:431)
com.databricks.logging.UsageLogging.$anonfun$withAttributionContext$1(UsageLogging.scala:239)
scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
com.databricks.logging.UsageLogging.withAttributionContext(UsageLogging.scala:234)
com.databricks.logging.UsageLogging.withAttributionContext$(UsageLogging.scala:231)
com.databricks.DatabricksMain.withAttributionContext(DatabricksMain.scala:74)
com.databricks.logging.UsageLogging.withAttributionTags(UsageLogging.scala:276)

The currently active SparkContext was created at:

org.apache.spark.SparkContext.<init>(SparkContext.scala:99)
com.databricks.backend.daemon.driver.DatabricksILoop$.$anonfun$initializeSharedDriverContext$1(DatabricksILoop.scala:347)
com.databricks.backend.daemon.driver.ClassLoaders$.withContextClassLoader(ClassLoaders.scala:29)
com.databricks.backend.daemon.driver.DatabricksILoop$.initializeSharedDriverContext(DatabricksILoop.scala:347)
com.databricks.backend.daemon.driver.DatabricksILoop$.getOrCreateSharedDriverContext(DatabricksILoop.scala:277)
com.databricks.backend.daemon.driver.DriverCorral.com$databricks$backend$daemon$driver$DriverCorral$$driverContext(DriverCorral.scala:180)
com.databricks.backend.daemon.driver.DriverCorral.<init>(DriverCorral.scala:217)
com.databricks.backend.daemon.driver.DriverDaemon.<init>(DriverDaemon.scala:39)
com.databricks.backend.daemon.driver.DriverDaemon$.create(DriverDaemon.scala:211)
com.databricks.backend.daemon.driver.DriverDaemon$.wrappedMain(DriverDaemon.scala:216)
com.databricks.DatabricksMain.$anonfun$main$1(DatabricksMain.scala:106)
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
com.databricks.DatabricksMain.$anonfun$withStartupProfilingData$1(DatabricksMain.scala:336)
com.databricks.logging.UsageLogging.$anonfun$recordOperation$4(UsageLogging.scala:431)
com.databricks.logging.UsageLogging.$anonfun$withAttributionContext$1(UsageLogging.scala:239)
scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
com.databricks.logging.UsageLogging.withAttributionContext(UsageLogging.scala:234)
com.databricks.logging.UsageLogging.withAttributionContext$(UsageLogging.scala:231)
com.databricks.DatabricksMain.withAttributionContext(DatabricksMain.scala:74)
com.databricks.logging.UsageLogging.withAttributionTags(UsageLogging.scala:276)
         
	at org.apache.spark.SparkContext.assertNotStopped(SparkContext.scala:132)
	at org.apache.spark.SparkContext.broadcast(SparkContext.scala:1617)
	at org.apache.spark.sql.execution.datasources.text.TextFileFormat.buildReader(TextFileFormat.scala:106)
	at org.apache.spark.sql.execution.datasources.FileFormat.buildReaderWithPartitionValues(FileFormat.scala:132)
	at org.apache.spark.sql.execution.datasources.FileFormat.buildReaderWithPartitionValues$(FileFormat.scala:123)
	at org.apache.spark.sql.execution.datasources.TextBasedFileFormat.buildReaderWithPartitionValues(FileFormat.scala:162)
	at org.apache.spark.sql.execution.FileSourceScanExec.inputRDD$lzycompute(DataSourceScanExec.scala:515)
	at org.apache.spark.sql.execution.FileSourceScanExec.inputRDD(DataSourceScanExec.scala:497)
	at org.apache.spark.sql.execution.FileSourceScanExec.doExecute(DataSourceScanExec.scala:719)
	at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:200)
	at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$3(SparkPlan.scala:252)
	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:165)
	at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:248)
	at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:192)
	at org.apache.spark.sql.execution.DeserializeToObjectExec.doExecute(objects.scala:104)
	at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:200)
	at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$3(SparkPlan.scala:252)
	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:165)
	at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:248)
	at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:192)
	at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:158)
	at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:157)
	at org.apache.spark.sql.Dataset.rdd$lzycompute(Dataset.scala:3247)
	at org.apache.spark.sql.Dataset.rdd(Dataset.scala:3245)
	at org.apache.spark.sql.execution.datasources.csv.TextInputCSVDataSource$.inferFromDataset(CSVDataSource.scala:151)
	at org.apache.spark.sql.execution.datasources.csv.TextInputCSVDataSource$.infer(CSVDataSource.scala:134)
	at org.apache.spark.sql.execution.datasources.csv.CSVDataSource.inferSchema(CSVDataSource.scala:72)
	at org.apache.spark.sql.execution.datasources.csv.CSVFileFormat.inferSchema(CSVFileFormat.scala:62)
	at org.apache.spark.sql.execution.datasources.DataSource.$anonfun$getOrInferFileFormatSchema$11(DataSource.scala:215)
	at scala.Option.orElse(Option.scala:447)
	at org.apache.spark.sql.execution.datasources.DataSource.getOrInferFileFormatSchema(DataSource.scala:208)
	at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:445)
	at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:384)
	at org.apache.spark.sql.DataFrameReader.$anonfun$load$2(DataFrameReader.scala:373)
	at scala.Option.getOrElse(Option.scala:189)
	at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:373)
	at org.apache.spark.sql.DataFrameReader.csv(DataFrameReader.scala:825)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:380)
	at py4j.Gateway.invoke(Gateway.java:295)
	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at py4j.commands.CallCommand.execute(CallCommand.java:79)
	at py4j.GatewayConnection.run(GatewayConnection.java:251)
	at java.lang.Thread.run(Thread.java:750)

[wajda]

I am not 100% sure it's directly related to that Spline plugin. Try the agent version 1.1.0 just to be sure.
Also, activate DEBUG log level to gather more details.

[zacayd]

when i moved to 1.1.0 it works