Hello, I am implementing Timeout for one of my flow. I have a few questions. 1. how to get the flow to enter

Timeout

state when timeout happens to a task. https://docs.prefect.io/api/latest/engine/state.html#timedout It seems there is only signals.FAIL instead of signals.TIMEOUT. 2. the fow DAG: A > B > C > D > E. Where E is always running with

trigger=triggers.always_run

. when timeout happens to task A, B, C, D, the task raises an exception

TimeoutError

, which is not a

FAIL

signal. and it causes the flow to terminate without running E properly. To overcome that, Currently I have to implement the exception handler in every task, it’s very tedious. Any better way to do it?

Try:
            except TimeoutError:
                raise signals.FAIL("Task Timeout.")

Happy Friday Prefect community! I just wanted to share this delicious flow run we recently had. I hope you're all loving Prefect as much as I am!

Hello, I ran into another issue with timeout setting. I would like to set a default timeout for all the tasks. I shall be able to do it through environment variable

PREFECT__TASKS__DEFAULTS__TIMEOUT

according to this doc

<https://docs.prefect.io/core/concepts/configuration.html#environment-variables>

However, when I do it properly, I got an error.

TypeError: Only integer timeouts (representing seconds) are supported.

even when I assigns the integer value. (See thread for details)

Hello again! When I set flow storage to GitHub, I don’t understand what the desired behaviour is supposed to be. 1. Does that mean that I do not need to keep my files locally? In a way that prefect automatically creates a temporary directory, pulls the scripts from the given repository with the credentials, executes the scripts and then deletes the script again? 2. Or do I still need to keep files locally? Does then prefect pull the tasks automatically? As far as I understand, I do not need to re-register the flow, but I can change the tasks and then push them. Plus I seem to have encountered a bug in the tutorial here: https://docs.prefect.io/core/idioms/file-based.html -> At the very end of the very first example, I ran into an error. I follow everything until I need to register the flow with this command:

prefect register flow -f flows/my_flow.py -p MyProject

This results in this error:

ValueError: Flow could not be deserialized successfully. Error was: ValidationError({'storage': {'ref': ['Field may not be null.']}})

Best wishes from Berlin!

Hi, I have deployed Prefect Server on K8s on GCP using the Prefect Server helm chart. Unfortunately, I'm experiencing some trouble when using the built-in agent that comes w/ the deployment. However, when I run

prefect agent kubernetes start

from outside the cluster w/in my own local machine's CLI, the error goes away and the flow runs successfully. Some insight on this would be much appreciated. I can confirm that the Prefect version of the deployed agent is correct too. Here's the error I'm getting from the using built-in agent that comes with the helm chart deployment:

[2020-12-05 10:30:03+0000] ERROR - prefect.CloudTaskRunner | Failed to set task state with error: ConnectionError(MaxRetryError("HTTPConnectionPool(host='prefect-server-gke-apollo.default', port=4200): Max retries exceeded with url: /graphql//graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f5a560d3690>: Failed to establish a new connection: [Errno -2] Name or service not known'))"))
Traceback (most recent call last):
  File "/opt/conda/lib/python3.7/site-packages/urllib3/connection.py", line 157, in _new_conn
    (self._dns_host, self.port), self.timeout, **extra_kw
  File "/opt/conda/lib/python3.7/site-packages/urllib3/util/connection.py", line 61, in create_connection
    for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM):
  File "/opt/conda/lib/python3.7/socket.py", line 752, in getaddrinfo
    for res in _socket.getaddrinfo(host, port, family, type, proto, flags):
socket.gaierror: [Errno -2] Name or service not known

Hi, We have noticed that some of our flows are running multiple times even if its just scheduled to run once a day. But the odd thing is its running exactly on the same time. Does anyone have similar issues? Also, I saw this https://github.com/PrefectHQ/prefect/issues/3758 just wanted to know if anyone knows if this has been resolved. Thank you!

Hello team, is it still not possible to kill a running task/flow from the cloud UI?

Hello friends, happy Monday. I’m getting a strange error in Prefect regarding

Failed to set task state with error: HTTPError('413 Client Error: Request Entity Too Large for url: <https://api.prefect.io/graphql>')

— whats the best way to debug this?

Hey 👋 I’m trying to figure out how to properly deploy Prefect on AWS ECS/Fargate such that we always have an agent running, ready to pick up any scheduled flow. I was able to execute

prefect agent fargate start -t $PREFECT_TOKEN

locally which appeared to work (it successfully started and was waiting for flows), but; 1. It never received any flows triggered from Prefect Cloud 2. This seems to be tied to my local terminal session (i.e. if I close my terminal, the agent shuts down) 3. I can’t see where this is actually deployed — I was expecting a new cluster to be created in AWS ECS, but that’s not the case Am I misunderstanding how this is supposed to work? If so, how can we go about deploying an always-running agent that’s ready to deploy new Fargate tasks whenever a flow is scheduled? Or is this not how Prefect is supposed to work either?

Hi folks, we’ve got some older flows that I’d like to get current with how Prefect works now, but I’m not entirely sure I understand how to fit our old paradigm into the new world. I’ll thread in because this might be long…

After having combed through a good amount of documentation and a handful of threads in this channel, I'm having a hard time understanding what an appropriate use case of a

Docker Task

is; it seems like if you want to string together a flow that pulls an arbitrary set of docker images and runs them as containers in some sequence, the canonical way to do this is to make each instance of pulling + running a container its own

Flow

(that leverages docker

storage

), and then to orchestrate a flow of flows. Given this, in what circumstance would you actually want to use a Docker Task? On the face of it, if you wanted to use one, it seems like you'd have to use some form of storage other than Docker for the task's parent flow to avoid a Docker-in-Docker situation; and I think not being able to use Docker Storage also implies that you wouldn't be able to use a Docker Agent for executing that flow + task? Does that sound right, or am I confusing the intended relationship between agent <-> flow <-> task?

Hi Community, I have a question about the prefect agents. If I have a prefect agent running on a machine can it run concurrent flows or does prefect need to have 2 agents running simultaneously to have concurrent flows running?

Hi Community, We have used graphql to extract prefect logs. The initial approach was just doing getting a list of flow, flow run and task run to iterate through the list. Obviously, it is the least performant way of extracting data from a database. We also started reviewing batching and caching techniques that Facebook uses with their dataloader. in the mean time, I am hoping to check in with the community experts and see if there is a much easier way to extract the logs without even going through graphql interface. Any info would be much appreciated.

Just arrived!

Looking to see if there is a definition of the functional API that can be used in the Flow context in a with statement. I was wondering why

with Flow('Arithmetic') as flow:
    x, y = Parameter('x'), Parameter('y')
    operations = {
        '+': x + y,
        '-': x - y,
        '*': x * y,
        '/': x / y
    }

resulted in operations containing {'+': <Task: Add>, '-': <Task: Sub>, '*': <Task: Mul>, '/': <Task: Div>} That is not what python would normally produce for the dict. "x+y" is coerced to <Task: Add>. "+" must be overridden in the Flow context. Where will I find the description of what magic is going on here? For instance, which operators are overridden?

https://docs.prefect.io/core/concepts/tasks.html#operators indicates that this happens, but not which operators are available. I am guessing that

+

is defined as something like this in the Flow Context:

lambda(x:Parameter, y:Parameter) Task("Add". lambda()(x.value + y.value))

Hi. Is it possible to run some flows with a concurrency limit and others without the limit using the same agent? It seem that if I start the agent with the label that has the concurrency limit it won't pickup flows that don't have that label.

Hi. I am noticing an inconsistency between the run time shown on the Flow page and that shown in the Flow run page. Screenshots and flow run ID in the thread.

Hi! I am running prefect on an on prem HTC cluster using 

HTCondor

 as workload manager and using a 

DaskExecutor

 connected to a cluster started with  

dask jobqueue

. I am mapping a function over a large number of images  (20000). I started running a subset of data (5000) using a predefine number of workers and the function is mapped and run correctly using all the workers available. If the cluster is started in adaptive  mode using 

dask jobqueue

  prefect is able to increase the number of processes run as expected by the adaptive mode and monitored in the workload manager however the size of the dask cluster doesn't change and the function isn't mapped, not even to the minimum number of workers that are predefined in the cluster. Interestingly 

HTCondor

 allocates new running processing but it seems to be independent from the dask cluster. It seems that the 

dask jobqueue

  initialised cluster cannot communicate with the processes started by prefect. After few minutes the processes started by prefect die printing out this error:

distributed.scheduler - ERROR - Workers don't have promised key: ['<tcp://192.168.0.2:40093>']
# the tcp address change

Any help will be greatly appreciated! Thanks!

Hi - we're seeing some duplicate scheduled flow runs in Prefect cloud. For example, this flow is supposed to run once a day, but we're getting duplicates.

Hi, is there a Prefect documentation page that list out all the supported Windows, Linux, MacOS platform versions? it is kind hard to go through trial and error to determine if Prefect will work smoothly with specific version of the OS's

Is there a way to run a task on a specific docker image, and not have every task in a run operate within the same docker container

I am reading in the prefect logs here that you can pull an image, start a container with a command, wait for the container to finish running, then stop the container. This is 4 tasks, but it would be really nice if I could just add a docker image name as an argument to the task decorator.

Hi guys, I have the following task to be automated. There a lot of CREATE TABLE SQL-scripts with dependecies (in FROM JOIN) between tables to be created and other long-living tables. These scripts are partially re-used from project to project. So in one project we use some part of these scripts, in onther project - another part. In each project we have some existing table which is used as input for these scripts (in FROM / JOIN part) And this code based growths. It looks like this codebase can be automatically (via SQL parser) represented as some DAG and in each project we can deal with subDAG parametrized by initial table. Is it good idea to use prefect as backend for modeling and executing such DAGs? We already implemented some prototype on Makefiles and actively use ability of GNU make to track dependencies between script's modification time and table creation time.. It makes possible to make only necessary work if some SQL-scripts are modified. As I understood such time-based execution model is not implemented in prefect. Am I right? Is it good idea to implement it manually? Does prefect have some support of Oracle database for results? Or maybe I should look on other solutions?

Hi, Looking at https://www.prefect.io/get-prefect/#pricing - Do "2 concurrent flow runs" (the Team plan) mean that only 2 flows can run at the same time? I'm aware that the question is self explanatory, however I'm a bit baffled that such a constraint is in place. Am I missing something?

Hi there, i am running into some memory utilization issues with some of my prefect flows, but am struggling to figure out why it seems memory isn’t being freed up or released between tasks in a specific flow: In the first graph here i have memory utilization for a specific flow (from cloudwatch metrics). As you can see memory steadily increases from 12:22 - 12:46 and then stays level until the flow completes. In the second attachment i have the flow and task durations during the flow. We can see that memory utilization steadily increases during the link patient events run, but after the task is completed the memory is never freed up. The flow in question is defined as such, so the link_patient_events tasks does not return anything other than a boolean: (Using prefect v0.11.5)

with SRMFlow(
        'TriggeredPatientLoaderFlow',
        feature_flag='patient_upsert_events_enabled'
) as trigger_patient_loader_flow:
    client_abbrev: str = Parameter("client_abbrev", default="")
    event_source_id: int = Parameter("event_source_id", default=None)

    lock_acquired: bool = acquire_event_source_lock(client_abbrev, event_source_id)
    linked: bool = link_patient_events(client_abbrev, event_source_id, lock_acquired)
    aggregated_patients: List[PersonInputProperties] = aggregate_patients(client_abbrev, linked, event_source_id)
    bulk_activity_metadata: dict = get_bulk_activity_metadata(client_abbrev, aggregated_patients)
    loaded: bool = load_aggregated_patients(client_abbrev, aggregated_patients, bulk_activity_metadata)
    marked_complete: bool = mark_event_source_as_completed(client_abbrev, event_source_id, loaded, lock_acquired)
    lock_released: bool = release_event_source_lock(client_abbrev, event_source_id, marked_complete, lock_acquired)

Is there a way to specify S3 storage option via config file (and avoid having to specify the bucket explicitly in each flow)?

Hi Prefect, after using graphql to update the flow name:

mutation {
  update_flow (_set: {name: "After Modified"}, where: {_and: {project_id: {_eq:"ea8106f2-1642-4e54-955a-3af1d7c7465e"}, name: {_eq: "Ori Flow Name"}} }){
    affected_rows
  }
}

it returned success and affected_rows=5 (5 different versions), but when I click 'start now' to create a new flow-run, it shows that:

Last State Message
[9 Dec 2020 9:26am]: Failed to load and execute Flow's environment: KeyError('After Modified')

Is there any step I missed? I just want to modify the flow name after I registered the flow. Thank you🙏

Hi I want to use SecretResult in my project. for learning purposes, I write an example project.

import requests
from prefect import Flow, task

from prefect.engine.results import SecretResult


@task(name='extract_task')
def extract():
    res = requests.get('<https://jsonplaceholder.typicode.com/todos/1>')
    res_txt = res.text
    print(res_txt)
    return res_txt


@task()
def load():
    """Print the data to indicate it was received"""
    print(SECRET_RESULT.location)
    res_txt = SECRET_RESULT.read('secret1')
    print("Here's your data: {}".format(res_txt))


with Flow("Callback-Example") as flow:

    SECRET_RESULT = SecretResult(extract, location='somewhere.json')

    e = extract()
    l = load()
    l.set_upstream(e)

flow.run()

but when I run it, gave some error on prefect source:

File "/home/hamed/PycharmProjects/etlprefectcloud/venv/lib/python3.8/site-packages/prefect/engine/results/secret_result.py", line 38, in read
    new.value = self.secret_task.run(name=location)
TypeError: extract() got an unexpected keyword argument 'name'

its on

prefect/engine/results/secret_result.py

line 38:

new.value = self.secret_task.run(name=location)

. and if I delete the name argumans, it works. am I using the SecretResult wrong way?