Hey Everyone! Quick question about Parameters. I have some tasks in which I’ve decided to access certain Parameter values via

prefect.context.parameters

. Am I correct to say that I can always rely on the context params being set before any other task is run in an async / Dask environment?

I'm working on a Selenium web scraping project, everything works fine when running locally - however, when I register the flow with Prefect Cloud, I'm not able to launch a Chrome Driver. It appears to be a serialization problem since I'm getting a

TypeError: cannot pickle '_thread.lock' object

. Has anyone run into a similar problem? Any suggestions would be appreciated!

How do I setup caching on a task that takes a python class instance as one of its arguments

The

PrefectResult()

result handler doesn't work since a class instance is not JSON serializable

If I use a serializer, are they any restrictions on what the class's attributes are? One of my classe's attributes is a function pointer

Hi folks! Has anyone had any experience using Sentry with the Prefect?

Hi community, I have a question around launching Agents and keeping them working. I understand, one of the most handy options is just using

nohup

, however, in order to do that, I have to execute everything with the

flow.run_agent()

option and then comment that line of code to execute the flows. The alternative is through CLI, however, I couldn’t make it work (it doesn’t file the libraries I need to execute, despite of using the

-p

param) Which is your experience/view here?

Hi everybody. We're using Prefect to run continuous processing on IoT data. I've just had to clear out the Prefect database as it grew to 12GB in less than a month, stalled the VM it was on, and I had no choice but to scrap it. I am logging a LOT of output from the continuous tasks, so my guess is that it's to blame for the growth, and can make sure once I have it running for a few days. The volume is a few log lines every 5 seconds from each of 5 continuous tasks. Regarding flow deletion/cleanup, I've found a 2 month old thread here https://prefect-community.slack.com/archives/CL09KU1K7/p1598535130019500 that mentions running a mutation on the Hasura API for doing deletions with a time filter, and here the suggested issue: https://github.com/PrefectHQ/server/issues/62 - @Dylan and @Sandeep Aggarwal you have participated in that thread. I have little experience with running GraphQL mutations and no experience with using the Hasura API, so I'm wondering if somebody has a working example of what to do here. Searches in Github and Slack for "clean"/"cleanup"/"database size" all give no results, so I'm hoping we can have a thread here with a recipe on what to do. I have no problem doing the legwork myself, for example, implement a Prefect flow that will make the needed API calls to cleanup old tasks. Any suggestions would be much appreciated.

Hey everyone 🙂

How do I check for the output of a function inside of flow?

I'm currently getting a None on print(realtor_data.result)

Just wondered what I should be using instead 👍

Hi prefecters 👋 I stumbled on an interesting case: how can I test (unit test) a task which fails and has a retry (but without waiting for all retries to go… ? Currently I do this, however this works only when all retires have passed (which in my case takes a long time). Is there a way around this?

def test_smth():
    with Flow("test") as flow:
        outcome = my_funct()
    state = flow.run()
    assert state.result[outcome].is_failed()

Hi folks, I'd want to configure logging for Rollbar. The issues is that the Rollbar auth token is available only at runtime in an environment variable. I was able to get the token using EnvVarSecret, which is then used to call rollbar.init(), but then I got stuck. How should I add the rollbar logger to prefect logging?

Hi there, I'm having some very tricky issues running tasks using

FargateAgent

. I am running this agent locally, and it seems to work intermittently. However, I am repeatedly getting this error:

[2020-10-22 17:59:48,212] INFO - agent | Starting FargateAgent with labels ['XXX']
[2020-10-22 17:59:48,212] INFO - agent | Agent documentation can be found at <https://docs.prefect.io/orchestration/>
[2020-10-22 17:59:48,212] INFO - agent | Agent connecting to the Prefect API at <https://api.prefect.io>
[2020-10-22 17:59:48,269] INFO - agent | Waiting for flow runs...
[2020-10-22 17:59:48,402] ERROR - agent | [{'path': ['get_runs_in_queue'], 'message': "'NoneType' object has no attribute 'flow_group_id'", 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

How can I register a flow from the command line? My setup:

# $ROOT/src/flow.py
from prefect import Flow

flow = Flow(name="test flow")

Now from the command line in dir

$ROOT

, I run:

prefect register flow --file src/flow.py --name "test flow"

I get the error

KeyError: "'__name__' not in globals"

What am I doing wrong? I’d like to use the CLI as it will be more simplistic from a CI/CD standpoint

Hi community! which is the best way to map over a dictionary? and acces the key and value on the mapping function, i could find a example with this!

Can anyone tell me how to kill all the scheduled Flows through the GUI? 🙂

So (finally) got it running, but the page doesn't load due to this error in Chrome :

auth0-spa-js must run on a secure origin

Which of course is completely ok because not running on localhost but rather on an ubuntu image and connect to

<http://192.168.88.133:8080>

in chrome address bar

<chrome://flags/#unsafely-treat-insecure-origin-as-secure>

added this domain. No luck, next error :

Refused to frame '<https://login.prefect.io/>' because an ancestor violates the following Content Security Policy directive: "frame-ancestors 'none'".

Anyone solved or run it on a different machine? Alternatively I end up running SSH and tunnel to it... Putty, ssh tunnel forward 8080 and 4200, and all is good. Maybe a nice addendum to the documentation.

i’m using prefect cloud - and running an agent via CLI locally. My agent is registered with cloud, but for the life of me I can’t figure out how I am supposed to register my flow. My bad understanding was that the registration communicated first with a proxy running on the local machine, but that doesn’t seem to be working. The error I see is:

prefect.utilities.exceptions.ClientError: Malformed response received from Cloud - please ensure that you have an API token properly configured.

What token am I supposed to configure? My agent has a runner token configured, but I don’t see where a flow has to have a token.

FYI this is the error I see whether using flow registration via the CLI or via

flow.register()

Hi all,

I am wondering how to make a

map

over a

DataFrame

, do you have any idea? I see very clear how to make it over a

list

. To my mind, it seems straightforward to convert the

DataFrame

into a

list

. So, I would like to know your opinion.

Has anyone ever seen an error like this during one of their prefect flows?

Failed to set task state with error: ClientError([{'path': ['set_task_run_states'], 'message': "can't handle event type ConnectionClosed when role=SERVER and state=SEND_RESPONSE", 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}])
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 111, in call_runner_target_handlers
    state = self.client.set_task_run_state(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 1177, in set_task_run_state
    result = self.graphql(
  File "/usr/local/lib/python3.8/site-packages/prefect/client/client.py", line 226, in graphql
    raise ClientError(result["errors"])
prefect.utilities.exceptions.ClientError: [{'path': ['set_task_run_states'], 'message': "can't handle event type ConnectionClosed when role=SERVER and state=SEND_RESPONSE", 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

I ran this flow 89 times with different input, and one time it failed with this message.

@nicholas After I saw that my flow failed last night, I tried to restart the flow right now through the Prefect UI. Originally, the flow failed after 15 minutes. I restarted it now (about 12 hours later) and it failed again. And now the Prefect UI says the flow's run time was >12 hours. That is really misleading and it screws with out metrics, since we monitor flow run time to keep track of how long tasks are taking.

Hi there, I am running into a very odd issue with regards to module packaging and registering to prefect cloud. The code lives here and the process to register lives here. Getting an

ModuleNotFoundError: No module named src

during the flow healthcheck, traceback here. Am I doing something wrong in terms of

__init__

packaging? This is a followup to thread yesterday.

I dont understand this example from the Docs; isn’t

flow

only defined inside that contexthandler/`with` statement?

from prefect import task, Flow

@task
def say_hello():
    print("Hello, world!")

with Flow("Run Me") as flow:
    h = say_hello()

flow.run() # prints "Hello, world!"

I’m getting the following

NameError: name 'flow' is not defined

For some reason my tasks are not running in parallel. Logs saying

Using executor type DaskExecutor

(i use local one).

import time

from prefect import task, Flow
from prefect.engine.executors import DaskExecutor
from prefect.environments import LocalEnvironment


@task
def wait_for(seconds):
    time.sleep(seconds)


with Flow('_Concurrency_') as flow:
    head = wait_for(5)
    leaf1 = wait_for(30)
    leaf2 = wait_for(15)
    tail = wait_for(3)

    head.set_dependencies(downstream_tasks=[leaf1, leaf2])
    leaf1.set_downstream(tail)
    leaf2.set_downstream(tail)

flow.environment = LocalEnvironment(executor=DaskExecutor())
flow.executor = DaskExecutor()

I expect leaf1 and leaf2 to run in parallel but they are not. In gantt chart i see they running sequentially. If i run them locally, they run in parallel. BTW whats the difference between environment executor and flow executor? Flows are run by docker agent.