prefect-community #prefect-community

prefect-community

Luan Tiburcio

05/26/2022, 12:58 PM

Good morning, I would like to know if there is a way to return a json directly through the graphql api in prefect1.0-ui??

Sang Young Noh

05/26/2022, 1:53 PM

Hello all I’m currently trying to run a code with a dask runner - I’ve set up a dask scheduler and client and tried running with something like the following code for the flow: I was just wondering if anyone was familiar with this error? Thanks

Jessica Smith

05/26/2022, 1:55 PM

can we get the logger name added to FlowRunLog? Not sure why it isn't there by default, it's kind of important

Stephen Lloyd

05/26/2022, 2:31 PM

I have a mapped task that will make an api call. What is the best way to count successful api responses? I could check the response message in the task and return an int along with the response. It seems like

send_tasks

will be a list of the responses and I could iterate over it in a subsequent task, but I’m hoping there is a more efficient method. Can I force a task to fail if the api indicates a failed call? If i do, is there a way to count those?

@task
def make_post(creds, task):
   ...
   response = requests.request('POST', ...)
   return response

with Flow(stuff, stuff):
   send_tasks = make_post.map(creds=unmapped(creds), task=data)

Will Skelton

05/26/2022, 3:28 PM

Hi All! I'm working on getting a Linux VM in Azure to host a prefect 2.0 server for POC/Dev purposes. I've followed the tutorial I've linked here but have gotten to a roadblock when trying to connect to the UI from my local computer. I've enabled Firewall rules successfully enough to allow SSH access into the vm via port 22 and have also added ports 8080 and 4200 as described in the tutorial. My issue comes when trying to bind the server to the public IP address for the vm. I've added the config.toml file as shown in the tutorial and have also tried to bind the IP using the "prefect orion start --host" command. Any thoughts on what I'm missing? Thanks! Will https://mortimer.xyz/prefect-on-azure/

Matthew Seligson

05/26/2022, 4:11 PM

When using the GitLab storage, in what scenarios is the flow code pulled? Is there caching? My assumption is that flow code is pulled from gitlab only upon initial start and restart from a failed state. Can the flow code be pulled while some tasks are running?

Kyle McChesney

05/26/2022, 4:31 PM

Is there any documentation around ResourceManagers and concurrency. Mostly just wondering about something like

with Flow(
    'resource_manager',
    executor=LocalDaskExecutor(),
) as flow:
    with ResourceManager() as manager:
        something_res = something(manager)
        get_res = another_thing.map(something_res, unmapped(manager))

Mostly wondering If I have the resource manager return “itself” how many copies are made via map

Edmondo Porcu

05/26/2022, 6:35 PM

Hi, are there examples of using https://docs.prefect.io/api/latest/tasks/aws.html#batchsubmit ?

Edmondo Porcu

05/26/2022, 6:37 PM

Also, if one uses a task that returns a list, can another task be executed for each element of the returned list?

Matthew Seligson

05/26/2022, 6:44 PM

I’d like to be able to run a flow from a set of start tasks. I understand this used to be supported but might not be any more. What’s the best way to go about this?

Renuka

05/26/2022, 6:53 PM

Is there a way to trigger databricks notebook from prefect 2.0 in the beta version ?

Patrick Tan

05/26/2022, 7:50 PM

prefect run command: How to I specify text parameter with non alpha numeric characters

Varun Srinivasan

05/26/2022, 11:15 PM

Hi.. I used https://gist.github.com/anna-geller/2014180ee5eaec9ea54f4d3f5b98ca93 as a reference to get exception logs.

class MyFlow(prefect.core.Flow):
    def __init__(flow_name, state_handlers, **kwargs):
        state_handlers = [log_stdout, on_flow_start, on_flow_finish]
        super().__init__(flow_name, state_handlers, **kwargs)


def on_flow_finish(flow: Flow, old_state: State, new_state: State) -> Optional[State]:
    if new_state.is_finished() and get_flow_state_name(new_state) == 'Failed':
        exception_trace: str = "".join(traceback.format_list(traceback.extract_tb(list(new_state.result.values())[0].result.__traceback__)))
        ## do something with exception_trace

While running it on prefect cloud, the

new_state.result

dict is empty resulting in a

IndexError

. When we run it locally, the new_result.result comes back with a length of the number of tasks in the flow. What's the right way to trap exceptions from a task in the flow's scope?

Dileep Damodaran

05/27/2022, 5:07 AM

Hi, I have been working with the prefect flows terminal state handler. From the terminal atate handler, Its possible to get only the state of every related tasks , but unable to identity the corresponding task details including task name. Is it possible to get all the reference tasks and its corresponding states to prepare a detailed summary of the flow run.

Joshua Greenhalgh

05/27/2022, 9:53 AM

Does anyone know of a way to apply some kind of tagging to the k8s jobs the agent starts so that I can easily use gcp metric explorer to see resource usage per flow_id? Everything is just

prefect-job-foo

not sure how I can isolate one flow from another?

Volker L

05/27/2022, 11:27 AM

Hi guys, I am running several python scripts once a day. Currently I am using crontab for the scheduling. I´d like to do this with prefect in the future using prefect orion without prefect cloud. Do anybone know of a simple tutorial or how to to get this up and running? Thanks 🙂

Tarek

05/27/2022, 11:51 AM

hi, i am new to prefect, i was able to run task in parallel using: flow.run(executor=LocalDaskExecutor()), but calling run() is blocking tell all tasks are finished, possible to call run in non-blocking way? so the run()-call returns immediately

Naga Sravika Bodapati

05/27/2022, 12:02 PM

any way to programmatically kill a flow that is running for more than a certain period of time? I am aware of SLA automations but want to do it code way. Please suggest. Thanks!

Thomas Opsomer

05/27/2022, 1:06 PM

Hola 🙂 I have an issue with a pod not terminating while the flow is finished. ⬇️

Shriram Holla

05/27/2022, 2:23 PM

Hi! I’m trying to run a flow locally with a prefect server. I have the server running as well as a local agent. When I run

flow.run()

, it errors out with this message:

TypeError: can't pickle _thread.lock objects

Note: I’m running the flow through a build system (bazel)

Jai P

05/27/2022, 3:59 PM

👋 hullo! curious, in prefect 2 is there a way to set some flow-level context that are accessible by all sub-flows/tasks?

Will Skelton

05/27/2022, 4:46 PM

Starting a new thread. Continuing from https://prefect-community.slack.com/archives/CL09KU1K7/p1653578937668889 We are working on running an orion server in an Azure VM. We seem to have gotten the UI accessible via a public IP and also have setup default storage (azure blob), added deployment have successfully run flows via the CLI. When we load the UI via the Public IP, the web page shows no information. Even the settings page fails to show the Orion version. I'm sure there is a setting (or access) somewhere that we are missing. Any thoughts?

✅ 1

Ilhom Hayot o'g'li

05/27/2022, 4:53 PM

Is that possible to use one task in multiple flows in order to follow "don't repeat yourself"? For example I have one task checker that will check difference of db to decide to update or not - that is used many times. So some code reuse maybe handy. Any suggestions? Some simple example, if difference_rows >0 : use task_to_update else : skip all tasks of this flow only

Kayvan Shah

05/27/2022, 4:53 PM

The extra logger that I have defined doesn't seem to work Then also tried overriding the logger defined in the in custom module Something strange happens here: Logs are visible in console but not on cloud or UI

Juan Valle

05/27/2022, 5:52 PM

Question regarding prefect server, we are running it in an EC2 Ubuntu instance with 20GB and getting close to using 100% of storage usage, is the server/docker containers storing log files? If so, where can I find out where?

Ilhom Hayot o'g'li

05/27/2022, 6:38 PM

Hi. I was trying to use conditionals instead of python if-else error occured 'reference tasks must be part of the flow'.

Amarpal Singh

05/27/2022, 6:51 PM

Has anyone else experiencing these errors with Prefect today? Capacity is unavailable at this time. Please try again later or in a different availability zone

Lin Lin

05/27/2022, 8:00 PM

Hello, everyone, I was getting the "something went wrong when you try to fetch the project". My setup is hosted prefect server on an EC2 box and on my local, I was trying to connect to the dashboard. I can see all the dashboard/menus, I can even list the project from the graghQL port, just can't list or create a project from the dashboard. Any idea where could go wrong?

Dharit Sura

05/27/2022, 8:06 PM

Hi Prefect Team!! I am trying to execute a deployment using Prefect 2.0 . I have created a default storage as Azure blob Storage and created a deployment. When I try to execute the deployment or run it through work-queue. I am receiving the below error. Currently I am testing this on windows OS. Please note that when I run ' python leo_flow.py ' I do see that the task and the flow is getting executed as expected. Any help or guidance around this would be highly appreciated!!

Edmondo Porcu

05/28/2022, 1:52 AM

Can we start using Prefect 2.0 with Prefect Cloud?