Channels
announcements
ask-marvin
best-practices-coordination-plane
data-ecosystem
data-tricks-and-tips
events
find-a-prefect-job
geo-bay-area
geo-berlin
geo-boston
geo-chicago
geo-colorado
geo-dc
geo-israel
geo-japan
geo-london
geo-nyc
geo-seattle
geo-texas
gratitude
introductions
marvin-in-the-wild
pacc-clearcover-june-12-2023
pacc-may-31-2023
ppcc-may-16-2023
prefect-ai
prefect-aws
prefect-azure
prefect-cloud
prefect-community
prefect-contributors
prefect-dbt
prefect-docker
prefect-gcp
prefect-getting-started
prefect-integrations
prefect-kubernetes
prefect-recipes
prefect-server
prefect-ui
random
show-us-what-you-got
Title
a
Alexandru Anghel
09/20/2022, 3:21 PMHi 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' objectFull error:
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' objectk
Kyle McChesney
09/20/2022, 3:25 PMWhat are your tasks inputs and outputs?
a
Alexandru Anghel
09/20/2022, 3:25 PM@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(
name=flow_name,
schedule=CronSchedule(cron=cron),
storage=GCS(bucket="prefect-bucket"),
run_config=KubernetesRun(
labels=["dev"],
job_template=register_flow_job_template,
service_account_name="prefect-sa",
# env={"PYSPARK_SUBMIT_ARGS": "--master <spark://spark-master-alerts:7077> --deploy-mode cluster pyspark-shell"},
image="prefect-base:v0.1",
image_pull_policy="Always",
)
) 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.
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-10.42.163.29-36489
22/09/20 15:16:25 INFO Master: Launching executor app-20220920151625-0000/1 on worker worker-20220920151515-10.42.87.199-43455k
Kyle McChesney
09/20/2022, 3:31 PMthanks 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
.toPandas()It might also just be the connection object. This seems related: https://github.com/PrefectHQ/prefect/discussions/3374
a
Alexandru Anghel
09/20/2022, 3:40 PMThanks 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