Hey all, Is there anyway to remotely remove/stop a local agent? A colleague has started a local agent with the same label as our regular agents and has been unreachable today to stop it so some of our flows are failing when sent there. I'm hoping there's a solution other than changing the label on all the flows. We're still on Prefect V1. Thanks 🙂

Hi! What is the proper way of scheduling a set of flows to run at for example midnight, but only one flow running at a time. So in a sequence (where order does not matter). I tried by just scheduling all the flows at the same time, one flow runs and the rest fails. I am running with Docker and the error message does not seem to be working properly:

Hi dear community, I am new at prefect, and I wanted to use the REST API with the prefect cloud. But the documentation does not specify which is the base URL and how to authenticate to it. Could you please clarify that for me. Thanks in advance

Hello. I am trying to apply deployment and I see that the whole project where I have a lot of flows will be uploaded to s3(my storage block). Why I need all this files there? I expect I have to upload only one file with flow code, right?

Hi everyone. We currently have a task which should take roughly 8 hours to finish running. This pipeline (wrapped in a task) is still in development and now and again it will not terminate. From what I can tell we can set a timeout in Prefect on the long running task but in this case we do not want to stop the task but rather send out a Slack notification to enable manual intervention. Is there a pattern/feature in Prefect 1.0 that allows for this?

Hi all, I’m getting error

No such command 'backend'.

Please help me fix this. Could this be because my prefect client and prefect server might be running different versions?

I’ve got a backfilling flow that is supposed to set off several thousand tasks, which ideally I’d have on the ConcurrentTaskRunner. When running locally on a powerful laptop this works fine, but on an Agent its CPU goes right to its limit but only a few tasks actually complete before it just stops doing anything. Is this to be expected at the minute or a surprise? Would this go better if I broke it up into lots of subflows instead of lots of tasks?

Hello Prefect Community, Great to be here! I am running a flow on prefect2 with a few tasks which have dependencies on local modules. We are running the tasks on a dask cluster and are using a s3 block storage. The issue I run into is recreating the local dependencies on the dask workers. Any suggestions on how you go about ensuring that the script environment is available to the dask worker?

Hello community, has anyone had issues implementing Prefect with Selenium? My case is really special and I can only use web scraping to get the data I need.

hey is there any “easy” tutorial on how to use Prefect 2.0 with AWS EKS, so far I haven’t found any and it’s kind of discovery mode for me.

Hi All, In the UI is it possible to somehow display the value that was mapped to the child task instead of

Mapped Child X

? I know I can get it from logs, but would be cool if I could see it in this screen

This message was deleted.

Howdy. We are using ORION and we have 2 agent servers running against the same worker queue. It appears they are both picking up the same jobs instead of delegating only 1. Anyone have something basic we might be missing?

Basic question about Prefect cloud. Please bear with me. So far have been using Prefect locally as follows: 1. pip install prefect 2. For the deployment, use a Python script inside which import Prefect and use the 'Deployment' 3. Then run 'prefect orion start' and it shows the scheduled runs and status at 127.0.0.1:4200 4. Then run 'prefect agent start' with queue name to serve the scheduled runs. Now looking at Prefect cloud, and have the following question: Should I do 'pip install prefect' anymore? If not and if prefect cloud takes care of it, how then would the Python program know about the 'Deployment', i.e. where do I import 'Deployment' from? Or should I stop the Python part altogether and schedule the deployment from Prefect Cloud? If yes, I presume that in the Prefect Cloud, I will have to specify the script that needs to be run at the scheduled time? In this case, my script will exist in an AWS EC2 instance, how do I give access to it to the Prefect Cloud? Thanks.

With Prefect 1.0, if I have multiple projects in my Prefect system... and I want to create a role that ... for instance, can only run flows based in Project A .. but not project B (Or even SEE Project B) where is that in the Customer Role section?

Hi All, I deployed the helm chart prefecthq/prefect-orion. For some reason, my pod is going into the CrashLoopBackOff state. Looking into the logs, I get this:

Error: No such command 'orion'.

I’m assuming this failure is coming from when it’s calling ‘prefect orion start’ Does anyone know what could be causing prefect to not recognize the orion command?

I created an API key in Prefect Cloud, and then I tried logging to the cloud account using the following command, and I get this error: (venv) D:\final>prefect cloud login --key <key-value> Unable to authenticate with Prefect Cloud. Please ensure your credentials are correct. Apart from configuring the API Key in the cloud, do I need to do anything else (in Cloud, or in the client machine) ? Thanks.

Good Afternoon! I have a question regarding using a snowflake query with wildcard

%

and pandas.read_sql. There seems to be an issue during compilation during a prefect flow run where it's complaining about the wildcard. We can run the queries outside of a prefect run no problem. When executing in prefect, we are getting formatting errors. Errors listed in threads

Im looking to implement a real-time (ish) pipeline in v1. An external system drops s3 files every minute or so. I'd like to poll s3 on a fixed-time-basis to see what files have been modified since the last poll. And if there are new files, grab them, and put them in the database. I used this as inspiration. What I did was first create "chunks" (or windows) to look at. So a list of n windows, with the tuple of the start/end time of the window. Then, mapping over each window, run a task that: (a) if the end of the window hasn't passed yet, throw RETRY signal with start_time=window_end_time, or (b) find the s3 files that were modified within the window. What seems to work pretty good, EXCEPT... when running the flow, the tasks that are in a pending state (from (a) above) are utilizing the available slots (flow is setup with num_threads=8). So essentially, my pipeline is not progressing to push data to a db until the last window has passed (not the intention).

Hi guys, I'am using

prefect 2.3.1

,

prefect-dask: 0.2.0

and got several errors running code similar to this pseudo code :

@task
def dask_context(item):
    name="myblock-name"
    config = await JSON.load(name)

@flow(task_runner=DaskTaskRunner())
def do_stuff():
   with tags("only_4"):
       result_list = dask_context.map(data_list)

I got this Exception multiple times after it running for a while (sometimes minutes, sometimes hours):

Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/httpcore/backends/asyncio.py", line 33, in read
    return await self._stream.receive(max_bytes=max_bytes)
  File "/usr/local/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 1265, in receive
    await self._protocol.read_event.wait()
  File "/usr/local/lib/python3.9/asyncio/locks.py", line 226, in wait
    await fut
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/httpcore/_exceptions.py", line 8, in map_exceptions
    yield
  File "/usr/local/lib/python3.9/site-packages/httpcore/backends/asyncio.py", line 35, in read
    return b""
  File "/usr/local/lib/python3.9/site-packages/anyio/_core/_tasks.py", line 118, in __exit__
    raise TimeoutError
TimeoutError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/httpx/_transports/default.py", line 60, in map_httpcore_exceptions
    yield
  File "/usr/local/lib/python3.9/site-packages/httpx/_transports/default.py", line 353, in handle_async_request
    resp = await self._pool.handle_async_request(req)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/connection_pool.py", line 253, in handle_async_request
    raise exc
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/connection_pool.py", line 237, in handle_async_request
    response = await connection.handle_async_request(request)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/connection.py", line 90, in handle_async_request
    return await self._connection.handle_async_request(request)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/http11.py", line 105, in handle_async_request
    raise exc
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/http11.py", line 84, in handle_async_request
    ) = await self._receive_response_headers(**kwargs)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/http11.py", line 148, in _receive_response_headers
    event = await self._receive_event(timeout=timeout)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_async/http11.py", line 177, in _receive_event
    data = await self._network_stream.read(
  File "/usr/local/lib/python3.9/site-packages/httpcore/backends/asyncio.py", line 35, in read
    return b""
  File "/usr/local/lib/python3.9/contextlib.py", line 137, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/local/lib/python3.9/site-packages/httpcore/_exceptions.py", line 12, in map_exceptions
    raise to_exc(exc)
httpcore.ReadTimeout

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/prefect/engine.py", line 1185, in orchestrate_task_run
    result = await task.fn(*args, **kwargs)
  File "/tmp/tmpq25d58myprefect/lib/dask_context.py", line 14, in _dask_task
    with await get_context(flow_run.parameters[block_name]):
  File "/tmp/tmpq25d58myprefect/./lib/config_context.py", line 40, in get_context
    config = await JSON.load(name)
  File "/usr/local/lib/python3.9/site-packages/prefect/blocks/core.py", line 606, in load
    block_document = await client.read_block_document_by_name(
  File "/usr/local/lib/python3.9/site-packages/prefect/client.py", line 1268, in read_block_document_by_name
    response = await self._client.get(
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1751, in get
    return await self.request(
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1527, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/usr/local/lib/python3.9/site-packages/prefect/client.py", line 257, in send
    await super().send(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1614, in send
    response = await self._send_handling_auth(
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1642, in _send_handling_auth
    response = await self._send_handling_redirects(
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1679, in _send_handling_redirects
    response = await self._send_single_request(request)
  File "/usr/local/lib/python3.9/site-packages/httpx/_client.py", line 1716, in _send_single_request
    response = await transport.handle_async_request(request)
  File "/usr/local/lib/python3.9/site-packages/httpx/_transports/default.py", line 353, in handle_async_request
    resp = await self._pool.handle_async_request(req)
  File "/usr/local/lib/python3.9/contextlib.py", line 137, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/usr/local/lib/python3.9/site-packages/httpx/_transports/default.py", line 77, in map_httpcore_exceptions
    raise mapped_exc(message) from exc
httpx.ReadTimeout

I started running this code with a tag concurrency limit of 20 then switched to 4. The more threads the exception was raised faster. Currently running on a server of 44 cores and 256 GB.

Hi - Prefect Experts, Im running into an error that indicates I am unable to schedule a flow that has required parameters. Is that expected? 🧵

Is there something that has to be done to my account to enable login through the CLI? I have created an API key but when I run

prefect cloud login

i get

Unable to authenticate with Prefect Cloud. Please ensure your credentials are correct.

Hi all, I was having issues with the logging in Prefect. I wanted to share my solution in case it helps someone else. Issue 1 - Prefect logger does not work outside of Prefect context Issue 2 - Prefect logger needs to have get_run_logger() in every functions Solution (proof-of-concept): See https://discourse.prefect.io/t/hybrid-logger-a-logger-that-works-both-inside-and-outsite-of-a-prefect-context/1559 @Taylor Curran

Is it possible to get a replica of the database of historical runs etc that Prefect cloud manages? Or something like incremental backups provided to end users

How do I add mapped tasks in the imperative API style? so for example:

flow = Flow("My imperative flow!")

# define some new tasks
name = Parameter("name")
second_add = add.copy()

# add our tasks to the flow
flow.add_task(add)
flow.add_task(second_add)
flow.add_task(say_hello)

# create non-data dependencies so that `say_hello` waits for `second_add` to finish.
say_hello.set_upstream(second_add, flow=flow)

# create data bindings
add.bind(x=1, y=2, flow=flow)
second_add.bind(x=add, y=100, flow=flow)
say_hello.bind(person=name, flow=flow)

If I wanted to give the

add

task a mapped input?

Hi all, I am getting this error when building a deployment..

Hi 👋 I am trying to use the GitHub repo feature that was added in 2.3.0 I have this code for creating the deployment (🔥*which doesn’t work*🔥):

from weekly_bonus_calc import weekly_bonus_calc
from prefect.deployments import Deployment
from prefect.filesystems import GitHub

storage = GitHub(
    repository="<https://github.com/yuvital/yuvital-prefect-github.git>"
)

deployment = Deployment.build_from_flow(
    flow=weekly_bonus_calc,
    name="weekly-bonus-calc",    
    work_queue_name="work-queue",
    storage=storage,
    schedule={'cron': '*/5 * * * *', 'timezone': "Asia/Jerusalem"}
)

deployment.apply()

Also, I’ve created a GitHub block via the UI (attaching a screenshot). Can you point out the correct way to setup this. Another thing I don’t understand is where are the Git credentials should go since this is obviously a private repo. Thank you for any help on the issue 🙏

hello everyone 👋, I am using prefect server 1.0 deployed on k8 with git as storage. I want to try event based triggering of prefect flows by a third party api , Is it possible in prefect server, if yes then how ?

hello everyone, I am using create_flow_run() task in a parent flow to trigger a flow.So when this flow failed in halfway.So my parent flow should restart it instead creating a new flow all together. but how to achieve the former case?