Hi! We use prefect to run databricks tasks, and we've been noticing lately that the status prefect shows for a task doesn't always update in time - prefect frequently thinks something is still running, or hasn't even started, when it already started and finished. So the flow doesn't continue on to the next tasks in the dag and gets stuck. Eventually gets the updated status of the databricks tasks and continues. This seems like some sort of communication issue between prefect and databricks, but I can't figure out how to debug. Would appreciate some help here! Thanks!

Are you mapping with the Databricks task by chance?

Sorry, not sure I understand what you mean

Like DatabricksSubmit.map(…) in Prefect

Got it - we're using DatabricksRunNow from the prefect tasks library

Ah ok. What executor are you using?

LocalDaskExecutor

Ah so do you do

DatabricksRunNow().map()

to parallelize?

we just do DatabricksRunNow() to run

I suspect you are doing something like this:

databricks_task = DatabricksRunNow()
with Flow() as flow..
     databricks_task(...)
     databricks_task(...)

Is that right?

with Flow('..') as flow:

flow.storage = GitHub(

repo="...",

path="...",

access_token_secret="..."

)

RunNowTask = DatabricksRunNow(databricks_conn_secret=json.loads(os.environ['<some auth related stuff>']))

RunNowTask.copy(

name=name,

job_id=job_id,

max_retries=max_retries,

retry_delay=retry_delay,

notebook_params=notebook_params,

)

RunNowTask()

flow.executor = LocalDaskExecutor(scheduler="threads", num_workers=4)

So there was this issue that the Databricks task was not thread safe

Cool - thanks for explaining! Will let you know if I'm still having this issue after the release. Appreciate the really fast responses here.

Of course!

Hi! Not sure if this is related, but now we are seeing tasks that fail with the message "Trigger was "all_successful" but some of the upstream tasks failed.". And when I look at the task runs, all upstream tasks have succeeded and prefect is aware of that.

Are you using Dask on Kubernetes?

Kubernetes

Is there a chance that your workers are pointed to Cloud? I’ve seen this before workers were reporting to Cloud instead of Server, causing the failure in the tasks

I'm not sure what exactly that means - how can I check?

Looking for a thread. Just to be clear, you are on a Dask Executor right? And these are mapped tasks?

Yes dask executor, but not mapped. Will check worker logs. Thanks!

I can’t find the thread I’m thinking of anymore. Let me just ask some questions before we go there. Does this happen all the time for this flow? Or does it succeed? Are you using your own image for Dask workers?

It sometimes works, sometimes fails - that's the confusing part for us

You can give me the code to how you spin up your DaskExecutor

We run the kubernetes prefect agent, doesn't that spin up the dask executor?

Not necessarily. It will run the Flow as a Kubernetes job but it’s the executor definition that will spin up the dask executor. The KubernetesRun can be LocalExecutor which will just take place on the job

I see. We don't have our own logic for spinning up the dask executor as far as I know

I understand what you are saying, but I am all the more convinced now. Could I see the flow snippet?

@Kevin Kho sorry for the delay - snippet is a little bit earlier in this thread history https://prefect-community.slack.com/archives/C014Z8DPDSR/p1634768730467400?thread_ts=1634767253.464300&cid=C014Z8DPDSR

Oh my bad. You might get a bit more stability with

flow.executor = LocalDaskExecutor(scheduler="processes", num_workers=4)

, but am still confused why this is happening

No worries! Can you please explain the difference between threads and processes for the scheduler?

That is the scheduler for the local dask cluster. This is a general concept rather than a Prefect/Dask concept. The processes are more isolated while the threads can share memory.

Thanks, I'll try and let you know how it goes

Ah I see. About heartbeats, read this

@Ritvik Mishra to reproduce the issue: 1. How long is your Databricks job running? 2. Which Prefect version do you use? 3. Which spark version do you use?

1. it wasn't running for very long when that happened, maybe an hour or so 2. not sure will check and get back to you 3. version 3.1.1

@Ritvik Mishra I investigated the issue in detail and I found a solution. In a default configuration, a flow triggering a Databricks job that runs for longer than 30 min keeps dying since Flow’s heartbeat is no longer detected with such long-running external jobs. For a Spark job that takes 30 min or less, the default configuration works fine. If you set the Flow’s heartbeat mode to “thread”, both

DatabricksRunNow

and

DatabricksSubmitRun

run until completion without any issues. I tested both tasks with a Databricks job that took an hour, and it worked well. You can set this mode using a single environment variable in your run configuration:

run_config=UniversalRun(env={"PREFECT__CLOUD__HEARTBEAT_MODE": "thread"})

I'm currently using KubernetesRun - will that env variable work still?

Yes the env var will work and then this may help you with the long running Databricks Job

Thanks will try it out!

can you share your run configuration and your Prefect version?

What do you mean by run configuration? Sent a lot of code snippets earlier in this thread, let me know if there is any info missing from there

prefect version

in the CLI for the second question. For the run configuration, you must have something like:

flow.run_config = KubernetesRun(..)

I think Anna wants to see how you included the env variable that she suggested.

flow.run_config = KubernetesRun( env={ "DATABRICKS_TOKEN": token, "PREFECT__CLOUD__HEARTBEAT_MODE": "thread", } )

Is that the version on the agent? or flow registration?

On the agent

For the Databricks task specifically, there were changes in 0.15.7 to help it become more stable. Changelog here . It might be worth upgrading to 0.15.7

Will try updating to 0.15.7