We have a custom ETL pipeline running as a python module on a Databricks cluster. At one point in this ETL pipeline we have to create a spark dataframe from multiple pandas dataframes. We do this with:
full_data = pd.concat(
[self.drift_data.null_proportions,
self.drift_data.categorical,
self.drift_data.numeric]
)
full_data_schema = StructType([ StructField("export_date", TimestampType(), True)\
,StructField("source", StringType(), True)\
,StructField("col", StringType(), True)\
,StructField("col_type", StringType(), True)\
,StructField("metric", StringType(), True)\
,StructField("category_value", StringType(), True)\
,StructField("val", DoubleType(), True)\
,StructField("threshold", DoubleType(), True)])
full_data_spark = self.spark_session.createDataFrame(full_data,schema=full_data_schema)
Most of the times we run this code on our cluster, we have no problems. However, sometimes when we run this code we get the following error:
PermissionError: [Errno 13] Permission denied: '/local_disk0/spark-6c598094-676d-4a43-bce1-b99b4bf36545/trustedTemp-254afa1a-ad7a-44b5-8992-dae14175fa3f/tmpma4wfpoy'
There does not seem to be any logical reason when this error pops up. We run the code on a high concurrency cluster with multiple user credential passthrough enabled. I have never seen this error arise when running on a standard cluster.
The error arise in the createDataframe line:
/databricks/python/lib/python3.8/site-packages/prm/drift_monitor/DriftMonitor.py in calculate_drift_data(self, date, source, dataframe)
343 ,StructField("threshold", DoubleType(), True)])
344
--> 345 full_data_spark = self.spark_session.createDataFrame(full_data,schema=full_data_schema)
346
347 #replace NaNs and nans from pandas to Spark NULLs
/databricks/spark/python/pyspark/sql/session.py in createDataFrame(self, data, schema, samplingRatio, verifySchema)
718 if has_pandas and isinstance(data, pandas.DataFrame):
719 # Create a DataFrame from pandas DataFrame.
--> 720 return super(SparkSession, self).createDataFrame(
721 data, schema, samplingRatio, verifySchema)
722 return self._create_dataframe(data, schema, samplingRatio, verifySchema)
I assume it has something to do with Pandas and permissions and the driver node but I can't seem to find anything helpful. Also the fact that it only happens (seemingly) at random has me left clueless.
Full error stack:
PermissionError Traceback (most recent call last)
<command-649024245350800> in <module>
1 prm = PRMP('dev')
----> 2 prm.run_platform()
/databricks/python/lib/python3.8/site-packages/prm/PRMP.py in run_platform(self, disable_date_check, perform_dropout_check)
50 ,wrapped_init=True)
51 for source in ['des', 'pop_prm_rem', 'pop_prm_pr']:
---> 52 dm.perform(source=source)
53
54 logger.info('finished PRM Platform run')
/databricks/python/lib/python3.8/site-packages/prm/drift_monitor/DriftMonitor.py in perform(self, source)
88
89 logger.info(f'starting metric calculation')
---> 90 self.calculate_drift_data(date=latest_export, source=source, dataframe=df)
91 logger.info(f'finished metric calculation')
92
/databricks/python/lib/python3.8/site-packages/prm/drift_monitor/DriftMonitor.py in calculate_drift_data(self, date, source, dataframe)
343 ,StructField("threshold", DoubleType(), True)])
344
--> 345 full_data_spark = self.spark_session.createDataFrame(full_data,schema=full_data_schema)
346
347 #replace NaNs and nans from pandas to Spark NULLs
/databricks/spark/python/pyspark/sql/session.py in createDataFrame(self, data, schema, samplingRatio, verifySchema)
718 if has_pandas and isinstance(data, pandas.DataFrame):
719 # Create a DataFrame from pandas DataFrame.
--> 720 return super(SparkSession, self).createDataFrame(
721 data, schema, samplingRatio, verifySchema)
722 return self._create_dataframe(data, schema, samplingRatio, verifySchema)
/databricks/spark/python/pyspark/sql/pandas/conversion.py in createDataFrame(self, data, schema, samplingRatio, verifySchema)
309 raise
310 data = self._convert_from_pandas(data, schema, timezone)
--> 311 return self._create_dataframe(data, schema, samplingRatio, verifySchema)
312
313 def _convert_from_pandas(self, pdf, schema, timezone):
/databricks/spark/python/pyspark/sql/session.py in _create_dataframe(self, data, schema, samplingRatio, verifySchema)
747 self._jvm.PythonSecurityUtils.SPARK_PYTHON_TO_JVM_TRUSTED_DATA_TRANSFER_ENABLED()) \
748 and not isinstance(data, RDD):
--> 749 jdf, schema = self._createFromLocalTrusted(map(prepare, data), schema)
750 else:
751 if isinstance(data, RDD):
/databricks/spark/python/pyspark/sql/session.py in _createFromLocalTrusted(self, data, schema)
560 # create the serializer as we do in self._sc.parallelize
561 serializer = self._sc.init_batched_serializer(data)
--> 562 temp_data_path = self._write_to_trusted_path(data, serializer)
563
564 return self._jsparkSession.createDataFrameFromTrustedPath(temp_data_path, schema.json()), \
/databricks/spark/python/pyspark/sql/session.py in _write_to_trusted_path(self, data, serializer)
541 temp_dir = self._jsparkSession.getTrustedPath()
542 # Create the temp file to dump data, this file would be deleted by the createDF call
--> 543 temp_file = NamedTemporaryFile(delete=False, dir=temp_dir)
544
545 # Serialize to the trusted path, still reclaim the handler if exception happens in dump
/usr/lib/python3.8/tempfile.py in NamedTemporaryFile(mode, buffering, encoding, newline, suffix, prefix, dir, delete, errors)
677 flags |= _os.O_TEMPORARY
678
--> 679 (fd, name) = _mkstemp_inner(dir, prefix, suffix, flags, output_type)
680 try:
681 file = _io.open(fd, mode, buffering=buffering,
/usr/lib/python3.8/tempfile.py in _mkstemp_inner(dir, pre, suf, flags, output_type)
387 _sys.audit("tempfile.mkstemp", file)
388 try:
--> 389 fd = _os.open(file, flags, 0o600)
390 except FileExistsError:
391 continue # try again
PermissionError: [Errno 13] Permission denied: '/local_disk0/spark-6c598094-676d-4a43-bce1-b99b4bf36545/trustedTemp-254afa1a-ad7a-44b5-8992-dae14175fa3f/tmpma4wfpoy'