Given a pyspark dataframe given_df, I need to use it to generate a new dataframe new_df from it.
I am trying to process the pyspark dataframe row by row using foreach() method. Lets say, for simplicity, both the dataframes given_df and new_df consists of a single column.
I have to process each row of this dataframe and based on the value present in that cell, I am creating some new Rows and adding it to new_df by unioning it with the Rows. The number of rows that are going to be generated upon processing a single row of given_df is variable.
new_df=spark.createDataFrame([], schema=['SampleField']) // Create an empty dataframe initially
given_df.foreach(func) // given_df already contains some data loaded. Now I run a function for each row.
def func(row):
rows_to_append = getNewRowsAfterProcessingCurrentRow(row)
global new_df // without this line, the next line will result in an error, because it will think that new_df is a local variable and we are trying to access it without defining it first.
new_df=new_df.union(spark.createDataFrame(data=rows_to_append, schema=['SampleField'])
However this results in a pickle error.
If the union function is commented out, then no error takes place.
PicklingError: Could not serialize object: Exception: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.
Traceback (most recent call last):
File "/databricks/spark/python/pyspark/serializers.py", line 476, in dumps
return cloudpickle.dumps(obj, pickle_protocol)
File "/databricks/spark/python/pyspark/cloudpickle.py", line 1097, in dumps
cp.dump(obj)
File "/databricks/spark/python/pyspark/cloudpickle.py", line 356, in dump
return Pickler.dump(self, obj)
File "/databricks/python/lib/python3.7/pickle.py", line 437, in dump
self.save(obj)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 789, in save_tuple
save(element)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/spark/python/pyspark/cloudpickle.py", line 500, in save_function
self.save_function_tuple(obj)
File "/databricks/spark/python/pyspark/cloudpickle.py", line 729, in save_function_tuple
save(state)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 859, in save_dict
self._batch_setitems(obj.items())
File "/databricks/python/lib/python3.7/pickle.py", line 885, in _batch_setitems
save(v)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 819, in save_list
self._batch_appends(obj)
File "/databricks/python/lib/python3.7/pickle.py", line 843, in _batch_appends
save(x)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/spark/python/pyspark/cloudpickle.py", line 500, in save_function
self.save_function_tuple(obj)
File "/databricks/spark/python/pyspark/cloudpickle.py", line 729, in save_function_tuple
save(state)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 859, in save_dict
self._batch_setitems(obj.items())
File "/databricks/python/lib/python3.7/pickle.py", line 885, in _batch_setitems
save(v)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 819, in save_list
self._batch_appends(obj)
File "/databricks/python/lib/python3.7/pickle.py", line 843, in _batch_appends
save(x)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/spark/python/pyspark/cloudpickle.py", line 500, in save_function
self.save_function_tuple(obj)
File "/databricks/spark/python/pyspark/cloudpickle.py", line 729, in save_function_tuple
save(state)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 859, in save_dict
self._batch_setitems(obj.items())
File "/databricks/python/lib/python3.7/pickle.py", line 885, in _batch_setitems
save(v)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 819, in save_list
self._batch_appends(obj)
File "/databricks/python/lib/python3.7/pickle.py", line 843, in _batch_appends
save(x)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/spark/python/pyspark/cloudpickle.py", line 500, in save_function
self.save_function_tuple(obj)
File "/databricks/spark/python/pyspark/cloudpickle.py", line 729, in save_function_tuple
save(state)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 859, in save_dict
self._batch_setitems(obj.items())
File "/databricks/python/lib/python3.7/pickle.py", line 885, in _batch_setitems
save(v)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 819, in save_list
self._batch_appends(obj)
File "/databricks/python/lib/python3.7/pickle.py", line 846, in _batch_appends
save(tmp[0])
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/spark/python/pyspark/cloudpickle.py", line 500, in save_function
self.save_function_tuple(obj)
File "/databricks/spark/python/pyspark/cloudpickle.py", line 729, in save_function_tuple
save(state)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 859, in save_dict
self._batch_setitems(obj.items())
File "/databricks/python/lib/python3.7/pickle.py", line 885, in _batch_setitems
save(v)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 819, in save_list
self._batch_appends(obj)
File "/databricks/python/lib/python3.7/pickle.py", line 846, in _batch_appends
save(tmp[0])
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/spark/python/pyspark/cloudpickle.py", line 500, in save_function
self.save_function_tuple(obj)
File "/databricks/spark/python/pyspark/cloudpickle.py", line 729, in save_function_tuple
save(state)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 859, in save_dict
self._batch_setitems(obj.items())
File "/databricks/python/lib/python3.7/pickle.py", line 885, in _batch_setitems
save(v)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 819, in save_list
self._batch_appends(obj)
File "/databricks/python/lib/python3.7/pickle.py", line 846, in _batch_appends
save(tmp[0])
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/spark/python/pyspark/cloudpickle.py", line 495, in save_function
self.save_function_tuple(obj)
File "/databricks/spark/python/pyspark/cloudpickle.py", line 729, in save_function_tuple
save(state)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 859, in save_dict
self._batch_setitems(obj.items())
File "/databricks/python/lib/python3.7/pickle.py", line 885, in _batch_setitems
save(v)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 859, in save_dict
self._batch_setitems(obj.items())
File "/databricks/python/lib/python3.7/pickle.py", line 885, in _batch_setitems
save(v)
File "/databricks/python/lib/python3.7/pickle.py", line 549, in save
self.save_reduce(obj=obj, *rv)
File "/databricks/python/lib/python3.7/pickle.py", line 662, in save_reduce
save(state)
File "/databricks/python/lib/python3.7/pickle.py", line 504, in save
f(self, obj) # Call unbound method with explicit self
File "/databricks/python/lib/python3.7/pickle.py", line 859, in save_dict
self._batch_setitems(obj.items())
File "/databricks/python/lib/python3.7/pickle.py", line 885, in _batch_setitems
save(v)
File "/databricks/python/lib/python3.7/pickle.py", line 524, in save
rv = reduce(self.proto)
File "/databricks/spark/python/pyspark/context.py", line 356, in __getnewargs__
"It appears that you are attempting to reference SparkContext from a broadcast "
Exception: It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.
For understanding a bit better what I am trying to do, let me give an example illustrating a possible use case :
Lets say given_df is a dataframe of sentences, where each sentence consist of some words separated by space.
given_df=spark.createDataframe([("The old brown fox",), ("jumps over",), ("the lazy log",)], schema=["SampleField"])
new_df is a dataframe consisting of each word at separate rows. So we will be processing each row of given_df and based on the words we get by splitting the row, we will be inserting each row into new_df.
new_df=spark.createDataFrame([("The",), ("old",), ("brown",), ("fox",), ("jumps",), ("over",), ("the",), ("lazy",), ("dog",)], schema=["SampleField"])
from Pickle error while creating new pyspark dataframe by processing old dataframe using foreach method
No comments:
Post a Comment