https://prefect.io logo
Channels
Powered by
j

Jovan Sakovic

01/28/2022, 6:41 PM
👋 Is there any known issues with Prefect and multi-threading? I’m having a difficult time trying to get the logging from the Thread’s target function to actually log in the UI.. Tried a bunch of stuff, and I thought I am now in the right place where I: • only make the logger objects within functions that are tasks (with 
logger = prefect.context.get("logger")
) • pass these logger object from function to function to finally get to the one that is running on threads ◦ 
extract_thread = Thread(target=extract_messages, name=f"Extractor #1", args=(extract_queue, load_queue, logger))
and 
extract_messages
looks like:
Copy code
def extract_messages(extract_queue, load_queue, logger):
    while True:
        request = extract_queue.get()
        
        if request == None:
            extract_queue.put(None)
            break

        request["extracted_at"] = datetime.now().strftime("%s")

        messages_request = api_request(request, logger)

        ## This never logs in the UI
        <http://logger.info|logger.info>("Queueing messages. ")
k

Kevin Kho

01/28/2022, 7:10 PM
If you are using multithreading, you can just use the LocalDaskExecutor instead. I don’t know exactly why, but in order to get this to work, you need to pass the logger explicitly in that function. I think the logger configuration doesn’t carry over to the thread
j

Jovan Sakovic

01/28/2022, 7:11 PM
Thanks, I’ll try the LocalDaskExecutor What do you mean by ‘pass the logger explicitly in that function’? Is that different than what I tried to do above?
And I guess a follow up here would be - any pointers on how to propagate an error that is raised within the same function that’s the target of a 
Thread
? If I raise an exception (
raise FAIL
)within that function nothing happens with the Flow. Would that also be taken care of if I implement the LocalDaskExecutor?
k

Kevin Kho

01/28/2022, 7:14 PM
Oh sorry I missed that. That should be enough. Yes it is taken care of with LocalDaskExecutor
Powered by