Hi everyone, I am trying to run a simple task aga...
# prefect-community
Hi everyone, I am trying to run a simple task against a spark cluster using Prefect 1.2 and i get this error:
TypeError: cannot pickle '_thread.RLock' object
I can see the application being registered inside the spark cluster. Thanks!
Full error:
Copy code
Unexpected error: TypeError("cannot pickle '_thread.RLock' object") Traceback (most recent call last): File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner new_state = method(self, state, *args, **kwargs) File "/usr/local/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 930, in get_task_run_state result = self.result.write(value, **formatting_kwargs) File "/usr/local/lib/python3.8/site-packages/prefect/engine/results/gcs_result.py", line 75, in write binary_data = new.serializer.serialize(new.value) File "/usr/local/lib/python3.8/site-packages/prefect/engine/serializers.py", line 73, in serialize return cloudpickle.dumps(value) File "/usr/local/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps cp.dump(obj) File "/usr/local/lib/python3.8/site-packages/cloudpickle/cloudpickle_fast.py", line 633, in dump return Pickler.dump(self, obj) TypeError: cannot pickle '_thread.RLock' object
What are your tasks inputs and outputs?
Copy code
@task(name='Get data', log_stdout=True, max_retries=2, retry_delay=timedelta(seconds=10), state_handlers=[post_to_slack_on_failure])
def get_data():
    spark = SparkSession.builder.appName("Bobita-app").master("<spark://spark-master:7077>").getOrCreate()
    return spark.createDataFrame([('look',), ('spark',), ('tutorial',), ('spark',), ('look', ), ('python', )], ['word'])

with Flow(
                # env={"PYSPARK_SUBMIT_ARGS": "--master <spark://spark-master-alerts:7077> --deploy-mode cluster pyspark-shell"},
        ) as flow:

             df = get_data()
@Kyle McChesney I've added above the task and flow snippets Logs from spark master proving the app is registered.
Copy code
22/09/20 15:16:25 INFO Master: Registered app Bobita-app with ID app-20220920151625-0000  
22/09/20 15:16:25 INFO Master: Launching executor app-20220920151625-0000/0 on worker worker-20220920151514-  
22/09/20 15:16:25 INFO Master: Launching executor app-20220920151625-0000/1 on worker worker-20220920151515-
thanks for sharing. Nothing sticks out as immediately wrong here. I think the error is related to trying to pickle the result from your task (the spark dataframe). I dont have a whole lot of spark experience, but is it possible that the spark dataframe is not pickle-able (ie, it might contain a reference to the session or connection which includes an RLock object)?
As a quick test maybe throw
on the end of the return
It might also just be the connection object. This seems related: https://github.com/PrefectHQ/prefect/discussions/3374
Thanks a lot, @Kyle McChesney. I'll have a look
Made some progress (i've added .toPandas() like you said), but now i get another error, this time inside the spark workers:
Caused by: java.io.IOException: Failed to connect to prefect-job-b3fe1959-274t6:42699
They are trying to connect to the Kubernetes job that started the flow. Maybe to return the state of the task? I am running the entire infra behind a corporate proxy and trying to figure out if this is causing the error.