Shoot, where did the TaskRunners move to? trying to update from Prefect 2.0b6 code to 2.0.2 and I cant find references to where they moved

Second question, I need to turn off checkpointing for a task, how would I do that in Prefect 2.0?

Me again, Thanks for that announcement of Prefect <> DBT. Could we add python typing in the docs (and in general)? For example, it's hard to understand what

result

is about ... Thanks

Is anyone using SSL connection for Postgres Database with prefect?

Hi all! In Prefect 2.0, what is the recommended way to have a task/flow call an existing docker image? The flow I'm writing takes some data, handles it with some tasks, feeds it to an external docker image, and then does something more with the output from the image. Been digging through the docs for a while, but I'm not really sure how to approach this. It's all running on my local machine. Thanks!

Hello all -- looking to see if anyone has a working Prefect 2.0 example of what it looks like to use a RemoteFileSystem block through the Cloud UI to setup SMB (including auth -- even if the auth can't be setup in the Cloud UI block itself) storage to use with flows deployment. Context: Our team doesn't have S3, GC, or Azure storage available to us. It all has to be kept on-premise. Our Data Center wants to provide us with SVM storage that's accessible via SMB (and it looks like fsspec does support a "smb://" connection string). Thanks!

I am trying to get my flow to run in a remote Ray (anyscale) cluster. However, my flow is failing whenever I try to submit a task. The error is that a Ray worker is trying to access something that is specific to my local machine (I am running directly on my local machine for development). I think this has something to do with local storage being used even in a distributed context. In the doc for RayTaskRunner, it recommends to use remote storage. I setup an S3 block, but then I have no idea what I suppose to do next. My flow is still using Local storage.

from prefect import flow, task, get_run_logger
from prefect_ray import RayTaskRunner


@task
def generate_numbers(n: int) -> list[int]:
    return list(range(n))


@task
def double_the_numbers(d: int) -> int:
    return d * 2


@flow(
    task_runner=RayTaskRunner(
        address="<anyscale://singularity/demo1>",
        init_kwargs={"runtime_env": {"working_dir": "."}},
    )
)
def run_workflow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello World!")

    numbers = generate_numbers.submit(100)
    <http://logger.info|logger.info>(numbers.result())
    doubled = double_the_numbers.map(numbers)

    <http://logger.info|logger.info>([d.wait().result() for d in doubled])


run_workflow()

ray::begin_task_run() (pid=20006, ip=<http://xxx.xx.xx.xxx|xxx.xx.xx.xxx>)
  File "/opt/homebrew/Caskroom/miniforge/base/envs/singularity2/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 193, in wrapper
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 140, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "/home/ray/anaconda3/lib/python3.10/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/home/ray/anaconda3/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/home/ray/anaconda3/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/home/ray/anaconda3/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/home/ray/anaconda3/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/engine.py", line 946, in begin_task_run
    return await orchestrate_task_run(
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/engine.py", line 1083, in orchestrate_task_run
    await _persist_serialized_result(
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/results.py", line 15, in _persist_serialized_result
    await filesystem.write_path(key, content)
  File "/home/ray/anaconda3/lib/python3.10/site-packages/prefect/filesystems.py", line 162, in write_path
    path.parent.mkdir(exist_ok=True, parents=True)
  File "/home/ray/anaconda3/lib/python3.10/pathlib.py", line 1177, in mkdir
    self.parent.mkdir(parents=True, exist_ok=True)
  File "/home/ray/anaconda3/lib/python3.10/pathlib.py", line 1177, in mkdir
    self.parent.mkdir(parents=True, exist_ok=True)
  File "/home/ray/anaconda3/lib/python3.10/pathlib.py", line 1177, in mkdir
    self.parent.mkdir(parents=True, exist_ok=True)
  File "/home/ray/anaconda3/lib/python3.10/pathlib.py", line 1173, in mkdir
    self._accessor.mkdir(self, mode)
PermissionError: [Errno 13] Permission denied: '/Users'

@Taylor Curran @Kalise Richmond

Hi, I was just curious to know how data is passed between tasks in prefect 2. Are the results always passed to the next task directly through memory even if storage is set up to persist them? Or is it just references to the storage that are passed between tasks? Thanks in advance

Hey all, we’re hosting @John Kang on PrefectLive today at 3p Eastern where he’s going to show how to use GCS as a storage for your Prefect workflow! Come join us live on Twitch!

I’m a little confused with the migration from v1 to Orion. Should we continue to use .map? Or write async tasks?

I set up an ECS agent that I can see running in prefect cloud, but when running a job I can see it is submitted but it will not execute. It is just stuck in the submitted state so not sure if I do not have enough storage of the agent can't get into one of the layers but there are not logs that say anything is wrong

Hi there, I have a CLI 2.0 question: is there currently a way to login to the Cloud API in a fully-scripted environment?

Hi! I’ve set up an ECS agent (Prefect Cloud 1.3.0) using this guide (great guide BTW). We’re using env variables to configure the used docker container. The problem is - newly spawned tasks in ECS do not get configured variables, and I do not see a way to explicitly forward them in the code. Is there a secure way to propagate env to flow runs other than specifying env in ECSRun (causing configuration duplication)?

Hello, is it just me or does anybody else also feel like Prefect 2 docs and UI are a downgrade compared to Prefect 1 docs and UI ? I feel Prefect 1 has much more polished and intuitive docs as well as the main UI. Is this because the main focus right now on Prefect 2 is to get the core framework right and then docs and UI will be upgraded and polished to bring it to Prefect 1 level ?

please help, prefect 2, when I try to run on a docker container

prefect cloud login --key...

it shows the following:

Hi all, what goes into "An Azure storage bucket path" when creating an Azure storage block? A couple of things I tried lead to "Container does not exist" errors from Azure. This is latest build of Prefect 2.0.2 with today's fix for .prefectignore included. (Yeah, I know... noob)

Hi all, trying to migrate my Prefect 1 GKE agent to Prefect 2 Orion and have it connect to Prefect Cloud. Followed the excellent demo on YouTube but when the pod starts up I get the following stack trace. I have tried 2.0.1 and 2.0.2 with different Python versions and I get the same error message back each time. Have also tried different ways to configure the kubectl secrets to see if that is the issue, wondering if others have run into the same issue?

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/cli/_utilities.py", line 41, in wrapper
    return fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 193, in wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 140, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "/usr/local/lib/python3.10/site-packages/anyio/_core/_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "/usr/local/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "/usr/local/lib/python3.10/site-packages/prefect/cli/agent.py", line 93, in start
    async with OrionAgent(
  File "/usr/local/lib/python3.10/site-packages/prefect/agent.py", line 249, in __aenter__
    await self.start()
  File "/usr/local/lib/python3.10/site-packages/prefect/agent.py", line 237, in start
    await self.default_infrastructure._save(is_anonymous=True)
  File "/usr/local/lib/python3.10/site-packages/prefect/blocks/core.py", line 618, in _save
    await self.register_type_and_schema(client=client)
  File "/usr/local/lib/python3.10/site-packages/prefect/blocks/core.py", line 560, in register_type_and_schema
    block_type = await client.read_block_type_by_slug(
  File "/usr/local/lib/python3.10/site-packages/prefect/client.py", line 1093, in read_block_type_by_slug
    return BlockType.parse_obj(response.json())
  File "/usr/local/lib/python3.10/site-packages/httpx/_models.py", line 743, in json
    return jsonlib.loads(self.text, **kwargs)
  File "/usr/local/lib/python3.10/json/__init__.py", line 346, in loads
    return _default_decoder.decode(s)
  File "/usr/local/lib/python3.10/json/decoder.py", line 337, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/local/lib/python3.10/json/decoder.py", line 355, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

Hi, we're running a prefect server locally (prefect 1.0) and we're facing some issues (attaching

docker ps -a

output)

3f832d699b5e   prefecthq/server:core-0.15.11
 this one occupied nearly 40G
42G	3f832d699b5edd87e39af62f724bb143826f2c5ae929a0565d0ceb479d2befde-json.log

• These logs are taking so much space that our root gets full and our local prefect server goes down, so is there any way we can remove old logs, or divide these logs in dated files and remove old ones as per need, or can we set size limit for these logs file ? • and also is there any command to get prefect server status and agent status, so that we can send some alerts as per server status and agents' status.

Hello everybody Pricing is being checked as version 2 comes out. In the previous version 1, the price increased in proportion to the number of task runs. However, when I checked the price policy of version 2, the contents of the number of task runs disappeared. Did the number of runs and the price become irrelevant after upgrading to version 2?

Hi all, We have Prefect Server (2.0) deployed and I'm looking into adding users. Is this possible in a standard free and self-deployed Prefect Server? I dont require more teams to be setup, the default one is sufficient but would like to add users. I'm assuming this isn't possible, but I see it's free up to 3 users in Cloud though

Hello everyone, I have a Prefect 1.0 Question: In this example from @Anna Geller https://github.com/anna-geller/packaging-prefect-flows/blob/master/flows/azure_kubernetes_run_custom_azure_image.py Why do we need to configure a storage in the first place, if we’re going to package our prefect flow and codebase in an image anyway (

image="<http://prefectdemos.azurecr.io/community/flows|prefectdemos.azurecr.io/community/flows>"

)? I understand that the default storage is local and if we don’t set up a storage then our kubernetes agent can fail, but I just want to understand why we need storage in the first place when we have a

KubernetesRun

Hello Guys! I have a couple of questions about prefect's scheduler and agent 1. Is it possible to make Flow run Instances Dependent on each other. By that I mean two things a. Trigger another Deployment based on the final state of a current Deployment Flow run instance. For example, if I have an ETL flow run that loads data to a database, if successful, i want it to trigger another flow that would read the data from this database and proceed with other logic. b. Making the Run of a Scheduled Deployment dependent on previous runs, so that if i have a Deployment scheduled to run every 15 mins, the 11:00 AM run, and other proceeding runs, would halt if the 10:45 AM run failed. 2. I've been running prefect for a PoC for my use-case and have been facing difficulties with the agent. It seems that the agent would suddenly crash and print out a large error message in the middle of a flow run. This is part of the error message

Traceback (most recent call last):
  File "/usr/lib/python3.8/asyncio/selector_events.py", line 848, in _read_ready__data_received
    data = self._sock.recv(self.max_size)
ConnectionResetError: [Errno 104] Connection reset by peer

From my understanding, this means that there was some connection problems with a TCP/IP connection and it wasn't able to send an acknowledgment message, however, I dont understand why it happens and it seems to me that it happens randomly (probably not true). Does it have to do with the size of data being transmitted (I am extracting data from an S3 bucket for processing)? Thanks in Advance!

Hello everyone! Question about Prefect 2.0. We started working on versions early 2.0b8 and in this version was possibility to submit Flow code with REST API with

flow_data

key, so we deployed flows this way and in our case it was very useful. We need to deploy flow with REST API without using Prefect cli or prefect as a python package. Prefect runs on a totally separate server. Now in 2.0.2 I cannot find how to do this. I read about blocks but I cannot find that block used as default in Prefect to store flow data. So can anyone explain how implement same flow as with

flow_data

in version 2.0b8 and early? How can I just submit Flow code with rest api to deploy it on Server? Thanks in Advance!

In Prefect 2.0 CLI, is there a way when running

prefect deployment ls

to get only the deployment’s ids without the fancy drawing of the table? (it makes it difficult to parse the output to use with

xargs

). Then another thing, I’ve noticed that when doing

prefect deployment apply

if the deployment already exists in the cloud, it doesn’t get updated. Is there a way to force the re-deployment of an existing one?

when i build prefect deployment with gcs storage it tries to copy all my files including the virtaul enviroment to the gcs bucket and then fails (venv) haris@haris-Inspiron-3593:~/Desktop/snowflake$ prefect deployment build workflow/venue_flow.py:gen_venue_flow -n genvenueflow_deployment -t n1-standard-4 -i kubernetes-job --storage-block gcs/prefect2-snowflake /home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/prefect/flows.py:176: UserWarning: A flow named 'gen_venue' and defined at 'workflow/venue_flow.py:6' conflicts with another flow. Consider specifying a unique

name

parameter in the flow definition:

@flow(name='my_unique_name', ...)

warnings.warn( Found flow 'gen_venue' Manifest created at '/home/haris/Desktop/snowflake/gen_venue_flow-manifest.json'. Traceback (most recent call last): File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/prefect/cli/_utilities.py", line 41, in wrapper return fn(*args, **kwargs) File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 193, in wrapper return run_async_in_new_loop(async_fn, *args, **kwargs) File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/prefect/utilities/asyncutils.py", line 140, in run_async_in_new_loop return anyio.run(partial(__fn, *args, **kwargs)) File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/anyio/_core/_eventloop.py", line 56, in run return asynclib.run(func, *args, **backend_options) # type: ignore File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 233, in run return native_run(wrapper(), debug=debug) File "/usr/lib/python3.8/asyncio/runners.py", line 44, in run return loop.run_until_complete(main) File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete return future.result() File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 228, in wrapper return await func(*args) File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/prefect/cli/deployment.py", line 502, in build file_count = await storage.put_directory() File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/prefect/filesystems.py", line 428, in put_directory return await self.filesystem.put_directory( File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/prefect/filesystems.py", line 250, in put_directory self.filesystem.put_file(f, fpath) File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/fsspec/asyn.py", line 86, in wrapper return sync(self.loop, func, *args, **kwargs) File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/fsspec/asyn.py", line 66, in sync raise return_result File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/fsspec/asyn.py", line 26, in _runner result[0] = await coro File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/gcsfs/core.py", line 1045, in _put_file await simple_upload( File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/gcsfs/core.py", line 1596, in simple_upload j = await fs._call( File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/gcsfs/core.py", line 392, in _call status, headers, info, contents = await self._request( File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/decorator.py", line 221, in fun return await caller(func, *(extras + args), **kw) File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/gcsfs/retry.py", line 115, in retry_request return await func(*args, **kwargs) File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/gcsfs/core.py", line 384, in _request validate_response(status, contents, path, args) File "/home/haris/Desktop/snowflake/venv/lib/python3.8/site-packages/gcsfs/retry.py", line 100, in validate_response raise ValueError("Bad Request: %s\n%s" % (path, msg)) ValueError: Bad Request: https://storage.googleapis.com/upload/storage/v1/b/prefect2-snowflake/o Payload size invalid. Expected size: 2. Actual size: 3. An exception occurred.

We are evaluating some solutions for our data workflows. For example, Google Cloud Composer (Airflow) and Prefect. If we decide to use Prefect, would be okay to starting using Prefect Cloud 2? It is still on Beta, when will be production ready?

Hey, Facing some issues with Orion 2.0.2 and not sure how to debug those. It seems to fail when calling

update_flow_run

in

engine.py

. I’m running a flow via the docker-container infrastructure type. It seems to start, but there’s no output on the UI, in fact the flow run status stays in “Pending” state, even after it fails on the work queue. I also see some TF log outputs which appear when importing TF, but the actual log output of my script is not showing up. Are there any hints on how to solve this?

We created a deployment:

<https://app.prefect.cloud/account/400***/workspace/8d8***/deployment/2f5***>

- it has the

staging

tag. We have a running agent:

$ /usr/bin/python3 /usr/local/bin/prefect agent start 360***
Starting agent connected to 
<https://api.prefect.cloud/api/accounts/400***/workspaces/8d8***>...
  ___ ___ ___ ___ ___ ___ _____     _   ___ ___ _  _ _____
 | _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
 |  _/   / _|| _|| _| (__  | |    / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|   /_/ \_\___|___|_|\_| |_|
Agent started! Looking for work from queue 
'360***'...

The work queue also has the tag

staging

. When I click on "Run" in the Cloud UI in the deployment, it is scheduled but the flow run has no tags assigned. And I guest this is the reason why it is never picked up by the agent. Or how can I debug this issue?

I'm trying to create a deployment with

prefect deployment apply deployment.yaml

but I keep getting an error from sqlalchemy:

sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) FOREIGN KEY constraint failed