Hi, I am doing some simple experiments with the caching system and ran into some issues. Here’s the code I am running:

from prefect.engine.results import LocalResult
from prefect import task, Flow

@task(target="add_target.txt", checkpoint=True, result=LocalResult(dir="."))
def add(x, y=1):
    ret = x + y
    return ret

@task
def print_value( x ):
    print( x )

with Flow("my handled flow!", result=LocalResult()) as flow:
    first_result  = add(1, y=2)
    second_result = add(x=first_result, y=100)
    print_value( second_result )

flow.run()

The

add

task only runs once and the final result is always 3. I read the code about

LocalResult

a bit and it seems to me that

LocalResult.exists

simply checks whether the target file exists. If so it will cause the task to enter cached state. So how can I get this example to work? Another issue is that even if I just have one

add

task in the flow, after modifying the source code to change its input, it still enters cached state. How can I make the result handler to check the task inputs as well?

Hi, I got the following error when fetching a bunch of images from a URL. What could be the possible reason for this? I am using prefect core. Unexpected error: ConnectionError(MaxRetryError('None: Max retries exceeded with url: /graphql (Caused by None)',),) Traceback (most recent call last): File "/usr/local/lib/python3.6/site-packages/prefect/engine/runner.py", line 48, in inner new_state = method(self, state, *args, **kwargs) File "/usr/local/lib/python3.6/site-packages/prefect/engine/flow_runner.py", line 598, in get_flow_run_state for t in final_tasks File "/usr/local/lib/python3.6/site-packages/prefect/engine/executors/dask.py", line 446, in wait return self.client.gather(futures) File "/usr/local/lib/python3.6/site-packages/distributed/client.py", line 1991, in gather asynchronous=asynchronous, File "/usr/local/lib/python3.6/site-packages/distributed/client.py", line 834, in sync self.loop, func, *args, callback_timeout=callback_timeout, **kwargs File "/usr/local/lib/python3.6/site-packages/distributed/utils.py", line 339, in sync raise exc.with_traceback(tb) File "/usr/local/lib/python3.6/site-packages/distributed/utils.py", line 323, in f result[0] = yield future File "/usr/local/lib/python3.6/site-packages/tornado/gen.py", line 729, in run value = future.result() File "/usr/local/lib/python3.6/site-packages/distributed/client.py", line 1850, in _gather raise exception.with_traceback(traceback) File "/usr/local/lib/python3.6/site-packages/prefect/engine/executors/dask.py", line 65, in _maybe_run return fn(*args, **kwargs) File "/usr/local/lib/python3.6/site-packages/prefect/engine/flow_runner.py", line 720, in run_task flow_result=flow_result, File "/usr/local/lib/python3.6/site-packages/prefect/engine/cloud/task_runner.py", line 52, in init self.client = Client() File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 124, in init tenant_info = self.graphql({"query": {"tenant": {"id"}}}) File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 281, in graphql retry_on_api_error=retry_on_api_error, File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 237, in post retry_on_api_error=retry_on_api_error, File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 401, in _request session=session, method=method, url=url, params=params, headers=headers File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 319, in _send_request response = session.post(url, headers=headers, json=params, timeout=30) File "/usr/local/lib/python3.6/site-packages/requests/sessions.py", line 581, in post return self.request('POST', url, data=data, json=json, **kwargs) File "/usr/local/lib/python3.6/site-packages/requests/sessions.py", line 533, in request resp = self.send(prep, **send_kwargs) File "/usr/local/lib/python3.6/site-packages/requests/sessions.py", line 646, in send r = adapter.send(request, **kwargs) File "/usr/local/lib/python3.6/site-packages/requests/adapters.py", line 516, in send raise ConnectionError(e, request=request) requests.exceptions.ConnectionError: None: Max retries exceeded with url: /graphql (Caused by None)

I am running into this failure state message:

Failed: At least one upstream state has an unmappable result.

It first came about after I included a switch statement that chooses between a mapped task, or a normal task that returns an iterable. For example:

with case(use_map, True):
    results = some_task.map(inputs) # some_task returns mapped results

with case(use_map, False):
    results = manual_mapping_task(inputs) # manual_mapping_task returns python iterable in its results

downstream_task(results) # "Failed: At least one upstream state has an unmappable result"

Is this a valid use of "results" in the downstream task? Is it possible this is where my error is coming from?

Hi 👋 can someone please assist with conditional flows? 🙂 https://github.com/PrefectHQ/prefect/discussions/3545

Hi 👋 I'm missing an option in `StartFlowRun`/`FlowRunTask` (old name) to schedule the flow into future. The change should be pretty straightforward, just passing

scheduled_start_time

here. Easy enough for me to start contributing 🙂 Are you interested in this addition or was there a reason behind not allowing this feature I'm not seeing?

Hello! I have a simple question, how can we pass a parameter from Prefect UI to a container ? We are using this code https://docs.prefect.io/core/examples/functional_docker.html but our job inside container take a parameter that we put in prefect UI

Hello all! Quick question, is it possible to run tasks inside a task? So for example:

@task
my_inner_task():
 # do something

@task
def my_task_one():
  my_inner_task()

Hi, Here's the sample code to the problem I mentioned yesterday were the prefect graph isn't parallelising as expected. We would expect the pre_processing function to be parallel but it isn't. The expensive computation function is being parallelised as expected. https://github.com/ralphwillgoss/prefect/tree/main/parallel

Hi, I sent an email because we want to upgrade our account to pay account "Your team is on the Prefect cloudCloud Developer plan. Contact us to change your plan or to get access to cool new features!"

Hi there, hope that someone could help me, I've got a Kubernetes Agent correctly installed, and I need to use Docker Storage for my flows, but i need to push the docker images on a private registry. My jobs fail with this error

Kubernetes Error: rpc error: code = Unknown desc = Error: image XXXXXX not found

So I've tried to tweak the job spec using

KubernetesJobEnvironment(_job_spec_file_="job_spec.yaml")

adding an

imagePullSecrets

to the Job template But the custom yaml that i defined seems not to be used when the job is spawned by the agent, what am I doing wrong? Here is how i register the flow:

flow.environment = KubernetesJobEnvironment(job_spec_file="job_spec.yaml")
      flow.storage = Docker(
        registry_url="REGISTRYURL",
        python_dependencies=['psycopg2-binary', 'pymssql'],
        )
      flow.register(args.register)

Hi Everyone! new here (obvious), first things first, thank you for running and managing a community!

Is this also the place for dropping config / first time install questions?

Im doing this clean install on Ubuntu 18, python 3.6 , steps:

new user: prefect , added to sudo and docker, nothing special regarding environment.

docker was already installed, I did add docker-compose

stopped my running postgres service, so it doesnt have any port conflicts, however would love to use this one.

pip install prefect

---> This creates a .local in my prefect home, navigate to /bin and /lib , took a closer look, run

prefect backend server

---> creates a .prefect with a .toml file with setting : backend = "server" , so far so good I guess. Now comes the next step in the documentation, run

prefect server start

---> Starts popping warning regarding postgres

WARNING: The PREFECT_SERVER_DB_CMD variable is not set. Defaulting to a blank string.
WARNING: The DB_CONNECTION_URL variable is not set. Defaulting to a blank string.
WARNING: The POSTGRES_DB variable is not set. Defaulting to a blank string.
WARNING: The POSTGRES_PASSWORD variable is not set. Defaulting to a blank string.
WARNING: The POSTGRES_USER variable is not set. Defaulting to a blank string.
Pulling postgres ... done
Pulling hasura   ... done
Pulling graphql  ... done
Pulling apollo   ... done
Pulling towel    ... done
Pulling ui       ... done
WARNING: The PREFECT_SERVER_DB_CMD variable is not set. Defaulting to a blank string.
WARNING: The DB_CONNECTION_URL variable is not set. Defaulting to a blank string.
WARNING: The POSTGRES_DB variable is not set. Defaulting to a blank string.
WARNING: The POSTGRES_PASSWORD variable is not set. Defaulting to a blank string.
WARNING: The POSTGRES_USER variable is not set. Defaulting to a blank string.
Creating network "prefect-server" with the default driver
Creating tmp_postgres_1 ... done
Creating tmp_hasura_1   ... done
Creating tmp_graphql_1  ... done
Creating tmp_towel_1    ... done
Creating tmp_apollo_1   ... done
Creating tmp_ui_1       ... done
Attaching to tmp_postgres_1, tmp_hasura_1, tmp_graphql_1, tmp_towel_1, tmp_apollo_1, tmp_ui_1
graphql_1   | bash: -c: line 0: syntax error near unexpected token `&&'
graphql_1   | bash: -c: line 0: ` && python src/prefect_server/services/graphql/server.py'
hasura_1    | {"type":"pg-client","timestamp":"2020-10-20T20:15:35.282+0000","level":"warn","detail":{"message":"postgres connection failed, retrying(0)."}}
hasura_1    | {"type":"pg-client","timestamp":"2020-10-20T20:15:35.282+0000","level":"warn","detail":{"message":"postgres connection failed, retrying(1)."}}
hasura_1    | {"type":"startup","timestamp":"2020-10-20T20:15:35.282+0000","level":"error","detail":{"kind":"db_migrate","info":{"internal":"could not connect to server: No such file or directory\n\tIs the server running locally and accepting\n\tconnections on Unix domain socket \"/var/run/postgresql/.s.PGSQL.5432\"?\n","path":"$","error":"connection error","code":"postgres-error"}}}
postgres_1  | Error: Database is uninitialized and superuser password is not specified.
postgres_1  |        You must specify POSTGRES_PASSWORD to a non-empty value for the
postgres_1  |        superuser. For example, "-e POSTGRES_PASSWORD=password" on "docker run".
postgres_1  |
postgres_1  |        You may also use "POSTGRES_HOST_AUTH_METHOD=trust" to allow all
postgres_1  |        connections without a password. This is *not* recommended.
postgres_1  |
postgres_1  |        See PostgreSQL documentation about "trust":
postgres_1  |        <https://www.postgresql.org/docs/current/auth-trust.html>

So where to start from here, adjusting docker compose file or other way to configure the local env?

Hello Everyone, Have started looking into prefect core library recently and exploring it to use for simple use cases. One thing I am struggling with right now is to provide a different timezone to prefect instead of default UTC. By default, prefect is generating UTC timestamps in logs but wanted to change it to my local timezone. Could you please help ?

So like to mention the following:

WARNING: The PREFECT_SERVER_DB_CMD variable is not set. Defaulting to a blank string.
WARNING: The DB_CONNECTION_URL variable is not set. Defaulting to a blank string.
WARNING: The POSTGRES_DB variable is not set. Defaulting to a blank string.
WARNING: The POSTGRES_PASSWORD variable is not set. Defaulting to a blank string.
WARNING: The POSTGRES_USER variable is not set. Defaulting to a blank string.

has to do with env vs export (shell vs bash), I've added added a

.env

file with the postgres var information and that actually worked when running

docker-compose up

searched it and it has to do with docker-compose. Not sure if its a bug in server.py Docker docker-compose environment vars Maybe someone is able to verify?

So with the Transform part of the ETL workflow, should there be 1 task for the transform that has multiple functions? or should there be multiple tasks? ex. If

data

is a list of records/dicts, and I want to process them into a csv.

with Flow('some flow') as flow:
   data = readData(source)
   transformedData = transformRecordsOneWay.map(data)
   transformedData = transformRecordsSecondWay.map(transformedData)
   transformedData = transformRecordsThirdWay.map(transformedData)

Hi folks, I am getting an INTERVAL_SERVER_ERROR trying to enable heartbeat using a graphql mutation query to prefect cloud

Hi all! Recently

task_run_name

was added to the

@task

decorator and it's beautiful -- however it prevents the flow from running locally for obvious reasons This makes local debugging (before deployment to Cloud) a pain since you'll have to remove and re-add this configuration all he time. Does anyone know a way to make Prefect just... ignore this parameter when the flow is run e.g. with

flow.run()

? Thanks!

Hi community! I have been exploring the

with case

possibilities and have the following question: • Using

with case

only worked under the

with Flow(...) as flow:

statement. When using

with case

within a

Task

, it didn’t work, does it make sense?

Thanks in advance!

Hi guys, need some help on daskexecutor - the agent is running on kubernetes and have registered the flow on cloud. I am using task result to mapped to another task: and i am getting following error:

Unexpected error: TypeError('Could not serialize object of type Success.\nTraceback (most recent call last):\n File "/usr/local/lib/python3.8/dist-packages/distributed/protocol/pickle.py", line 49, in dumps\n result = pickle.dumps(x, **dump_kwargs)\nTypeError: cannot pickle \'SSLContext\' object\n\nDuring handling of the above exception, another exception occurred:\n\nTraceback (most recent call last):\n File "/usr/local/lib/python3.8/dist-packages/distributed/protocol/serialize.py", line 258, in serialize\n header, frames = dumps(x, context=context) if wants_context else dumps(x)\n File "/usr/local/lib/python3.8/dist-packages/distributed/protocol/serialize.py", line 61, in pickle_dumps\n frames[0] = pickle.dumps(\n File "/usr/local/lib/python3.8/dist-packages/distributed/protocol/pickle.py", line 60, in dumps\n result = cloudpickle.dumps(x, **dump_kwargs)\n File "/usr/local/lib/python3.8/dist-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps\n cp.dump(obj)\n File "/usr/local/lib/python3.8/dist-packages/cloudpickle/cloudpickle_fast.py", line 563, in dump\n return Pickler.dump(self, obj)\nTypeError: cannot pickle \'SSLContext\' object\n')

Cant we serialize output of a task and use daskexecutor to run parallel mapped tasks?

set task nameHi all, a question about graphql client timeouts arbitrarily happening when running on fargate with

Hi, When using the LocalDaskExecutor and specifying the number of workers, will each worker use one thread or as many as it can? Will prefect map one task to one worker then?

Hi 👋 🙂 Is it possible to set a task which already ran successfully as RETRY from a job down the stream? My situation is this: I have data which i want to write to a DB. To do this I have 3 tasks for the DB: •

open_db_connection

(run only once during a flow) •

save_to_db

run multiple times during one flow •

close_db_connection

- run once at the end of the flow Sometimes when I

save_to_db

the connection has dropped out (for various reasons). Would a clean solution be to set the current

save_to_db

task as

RETRY

and then set one of the tasks it depends on (the

open_db_connection

task) as RETRY too. This way it will (or so I expect) to open a db connection first and then retry the failed task again. 🤔 is my thinking correct and is that possible or it’s a messy solution? Thanks : )

Hi, I just saw this issue https://github.com/PrefectHQ/prefect/issues/3058 and it sounds an awful lot like the reason why my flow is failing. However, I did not understand in what way it was resolved. Is adding

"<http://cluster-autoscaler.kubernetes.io/safe-to-evict|cluster-autoscaler.kubernetes.io/safe-to-evict>": "false"

currently the way to go? Other than that, we currently use the Helm charts from https://github.com/PrefectHQ/server/pull/57 which work pretty great so far. Is there a way to assist to push that forward since "official" Helm charts would be pretty awesome (however I don't have a lot of experience with Helm).

Another small question: the docs for the Dask Kubernetes Environment show an example where they set

--memory-limit 4GB

in the

dask-worker

command but also set a memory limit of

memory: 4G

in the Kubernetes

resources

. Are both required for correct functionality?

In https://github.com/PrefectHQ/prefect/issues/470 throttling was removed and it says that “We will replace it with a robust, global solution.” Has there been any replacement or progress on implementing any sort of flow throttling? Specifically I’m looking to limit the number of concurrently running flows based on certain criteria.

Hey there, looking for some ideation. My team has set up a monolithic repo for our various data related flows, attached with a CircleCI workflow to handle registering to Prefect Cloud. I was wondering - has anyone set up a CI/CD with multiple users? How were you able to register flows for a given user via CI/CD? Have people created service-level User accounts that consolidate all flow registration? Thanks!

I am wondering if anyone has ever experienced a "reduce" bottleneck using prefect. I have a procedure which consumes thousands of mapped tasks. The task itself is very simple, but it just hangs for a very long time. Does anyone have leads for what might be the issue.

I have a very basic question. What exactly is a "tenant" and how does the concept differ from a "user"?