👋 I don't understand how to create a deployment where the code is in a subdirectory in GitLab/Github

my_project.git
/src/
/src/dbt/
/src/dbt/Earthfile # Equivalent to Dockerfile
/src/dbt/my_prefect_job.py
/src/dbt/...

And my Earthfile is like

FROM prefecthq/prefect:2.7-python3.10
WORKDIR /code
...
COPY . .

And so

my_prefect_job.py

is under

/code/my_prefect_job.py

in my docker image and under

/src/dbt/

in my gitlab repository But once deployed, the run output the following

Downloading flow code from storage at 'src/dbt'

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 879, in exec_module
  File "<frozen importlib._bootstrap_external>", line 1016, in get_code
  File "<frozen importlib._bootstrap_external>", line 1073, in get_data
FileNotFoundError: [Errno 2] No such file or directory: '/code/prefect_job.py'

And so I don't understand why it cannot find my flow in gitlab under the right subdirectory nor the flow in the docker contain under the right path 🤷 Did not find any good example about github/gitlab subdir storage strategy. Thanks in advance 🙏

Is there a purpose to saving blocks for infrastructure? I use

KuberenetsJob

blocks to run flows on k8s - is there a benefit to saving rather than having what looks to be a temporary block called

anonymous-<UUID>

?

Hi I am encountering "No such file or directory" while using github as a storage. I am using prefect version: 2.7.9 and agent is running on a remote machine.

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 879, in exec_module
  File "<frozen importlib._bootstrap_external>", line 1016, in get_code
  File "<frozen importlib._bootstrap_external>", line 1073, in get_data
FileNotFoundError: [Errno 2] No such file or directory: '/tmp/tmppy44ikxrprefect/sample_flow.py'

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

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 266, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/usr/local/lib/python3.10/site-packages/prefect/client/utilities.py", line 47, in with_injected_client
    return await fn(*args, **kwargs)
  File "/usr/local/lib/python3.10/site-packages/prefect/deployments.py", line 186, in load_flow_from_flow_run
    flow = await run_sync_in_worker_thread(load_flow_from_entrypoint, str(import_path))
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 91, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(
  File "/usr/local/lib/python3.10/site-packages/anyio/to_thread.py", line 31, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 937, in run_sync_in_worker_thread
    return await future
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 867, in run
    result = context.run(func, *args)
  File "/usr/local/lib/python3.10/site-packages/prefect/flows.py", line 772, in load_flow_from_entrypoint
    flow = import_object(entrypoint)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/importtools.py", line 195, in import_object
    module = load_script_as_module(script_path)
  File "/usr/local/lib/python3.10/site-packages/prefect/utilities/importtools.py", line 158, in load_script_as_module
    raise ScriptError(user_exc=exc, path=path) from exc
prefect.exceptions.ScriptError: Script at 'sample_flow.py' encountered an exception: FileNotFoundError(2, 'No such file or directory')

Here is the deployment file:

from sample_flow import start_sample_flow
from prefect.deployments import Deployment
from prefect.filesystems import GitHub

deployment = Deployment.build_from_flow(
    flow=start_sample_flow,
    name="sample-pipeline",
    version="2",
    tags=["VA_PIPELINE"],
    storage=GitHub.load("dev"),
    skip_upload=True,
    infra_overrides={"env": {"PREFECT_LOGGING_LEVEL": "DEBUG"}},
    path="cloud/prefectflows"
)

if __name__ == "__main__":
    deployment.apply()

any help here is much appreciated

Crosspost from #prefect-ui https://prefect-community.slack.com/archives/C0192RWGJQH/p1676392806971479

Any time I try to write unit tests for a flow with mapped tasks, I am getting a

MappingLengthMismatch

with the following traceback

Parameters for map must all be the same length. Got lengths: {'args': 0, 'kwargs': 2}

However, all of the parameters for the task do have the same length

is it possible to create Automations using the GraphQL API? This seems to suggest there is; however it seems like you can only create actions, and not Automations. The docs don't return anything for

automation

Hi Prefect team! We have been seeing an increased number of instances when working with asynchronous flows where, for one reason or another, the flow begins to hang. This typically occurs in one of two instances... 1. Call from inside async flow to sync task 2. Call from inside async flow to async task I was wondering what the expected behavior in these two cases would be? I see that with 2.0.2 an almost mirrored version of this issue was patched. Thank you in advance!

Hi Prefect team, my company is migrating from Prefect1 to Prefect2. We have a few Prefect1 flows with Great Expectation integrations. But I haven’t found much information on how GE works with Prefect2. Could someone please share some information? Or is it just the same as in Prefect1?

Is there a limit ( / rule-of-thumb) to the number of subflows created from a flow?

Hi everyone. I have just started looking into prefect and coiled and planning to use these 2 with ECS Cluster. Does anyone have any recommendations about good tutorials, examples, scripts on those? I have watched most from YouTube and looking into documentations on Coiled and Prefect pages now, but didn't find clear workaround on these.

Hello everyone, Just wanted to follow-up in case the message was missed. We are still facing the same issue and it is spreading to other flows now. Your help is appreciated 🙇 🙇

Hi all, a quick one about running prefect agents on windows, with enterprise cloud server. I am facing an issue where after long-running tasks complete the agents just stop. The last flows exit cleanly, and the next message I get is just

Agent stopped!

- is there any reason why agents would just be stopping? I thought they were meant to run indefinitely. I do not know where to access any logging to see why the agent was stopped.

Hi guys. When I build and apply a deployment I'm getting error below. Storage block is not set, thus local storage is used. Infra block specifies k8s cluster and an agent works in there, so basically I want to run a flow from my local storage in k8s cluster specified by infra block. Any idea?:

Is it possible to somehow set the --memory= flag for the docker block to limit the memory usage of the flow runs per container?

Hi, quick question - what would be the best/idiomatic way to run a flow repeatedly? ie. as soon as it completes, start it again.

I've got a few datasets that are "partitioned" in both the traditional sense of dates, but also in the sense of regular "dump files" that I'd like to EL into a database. How, in practice, do folks recommend treating such paritioned datasets in Prefect?

Hi team! Does anyone know when prefect 1.0 get discontinued? Or deprecated?

Some deadline or similar?

Is there a way in prefect 1.0 to have task/flow concurrency and if other scheduled runs get submitted but the limit of concurrency is reached, instead of queueing just cancel those runs. thanks.

hi it looks like my prefect runs are continuing to duplicate in the ui (and I have about 160,000 runs). I would like to be able to delete these flow runs either through the prefect cli or through the python api. the delete command for prefect only deletes one at a time- and there is no select all option in the prefect ui. any suggestions on how to do this efficiently would be helpful!

Hello everyone! I'm experimenting with prefect2.0 and I ran in a situation I don't know how to address, hope someone can help me here. I'm downloading a bunch of CSV files, save them locally, and upload them to BigQuery one by one. Since data is quite large, I'm trying to skip the download task if the file has already been downloaded previously. Prefect1 had the

target

functionality that seemed to fit my case, but it doesn't exist in prefect 2.0. I tried in so many different ways but I can't figure out how. Reason I'm saving locally as an intermediate step is because I run OOM otherwise. My code (simplified) goes like this:

@task
def download(url, filename):
    file_path = f'/absolute/path/to/{filename}.csv'
    with open(file_path, 'wb') as file:
        res = requests.get(url+filename)
        file.write(res.content)
    return file_path

@task
def write_to_bq(file_path):
    df = pd.read_csv(file_path)
    df.to_gbq(...)

@flow
def download_all(url_list):
    paths = []
    for url in url_list:
        file_path = download(url)
        paths.append(file_path)
    return paths

@flow
def upload_all(paths):
    for path in paths:
        write_to_bq(path)

@flow
def main(url_list):
    paths = download_all(url_list)
    upload_all(paths)

Hi everyone. I was examining this example: https://github.com/PrefectHQ/prefect-recipes/blob/main/devops/github-actions/general-docker-deploy.yaml. This action builds an image and deployment. I couldn't see where image is used, it is just built and pushed to the registry. Thanks. 🙏

Hi all -- a fun one. I have noticed some of my flows failing with this error:

Task run '511abff8-faa3-4efc-94e6-f4be435db16e' received abort during orchestration: This run cannot transition to the RUNNING state from the RUNNING state. Task run is in RUNNING state

As far as I can tell it is inconsistent (work flow sometimes works, sometimes does not). I am running a

DaskTaskRunner

back by a

SLURMCluster

. The stage that is crashing is trying to read in a set of large-ish files, and I believe the GIL is not being released as the data is being accessed. At the moment there is a high load on the disk I/O, and my best guess is that the dask nanny is somehow failing a health check, and prefect in turn is causing some round-about error like this. Any ideas that make more sense?

We are running prefect server on Azure Kubernetes, all seems to be working fine except this error pop comes when we launch server UI. Any idea why https://<prefect_ui_URL>/api gives 404?

Can't connect to Orion API at https://<url>/api. Check that it's accessible from your machine.

Hi. Does kubernetes block has a configuration to differentiate between GPU and CPU nodes? Assume I want to run my training flows on GPU nodes, so is it configurable via blocks or should we add some taints and tolerations to the deployment yaml built by Prefect?

Hi, I am trying to run another flow(using run_deployment), is there is a way to track the status of it, I am not finding that in the documentation: https://docs.prefect.io/api-ref/prefect/deployments/#prefect.deployments.run_deployment

hello community, hello Prefect team, i am using the Ray task runner with a cluster that consists mainly of AWS spot instances. i would love to just restart tasks when the underlying spot instance is unexpectedly terminated. this functionality is already supported by Ray but Prefect's

PreventRedundantTransitions

orchestration policy prevents a repeated task execution as the task remains in status

Running

when the instance is terminated and a new replacement instance will try to start the task from scratch, including proposing the state

Running

again which leads to:

Task run <task_run_id> received abort during orchestration: This run cannot transition to the RUNNING state from the RUNNING state. Task run is in RUNNING state

I feel like this policy has a legitimate reason of preventing multiple agents from attempting to orchestrate the same run but also catches the case where the same agent tries to execute the same task multiple times which should be allowed imo. Anyone having experience with Prefect + Ray on AWS spot and ideas on how to handle this case? PS: @Tim Galvin not sure about Dask and SLURM but maybe your issue is somewhat related

Hello awesome people!! So I'm currently using prefect to orchestrate an ETL pipeline for a job take-home task and I was wondering if there's ANY way, I mean ANY way to make prefect work with command line arguments 😭😭

I already built out the entire project with this assumption in mind and I hadn't verified (terrible of me). I just need this feature to tie the whole thing together, any help would be appreciated

🚨 we're live! 🚨 come join Noteable, Prefect + former Prefect community engineer @Kevin Kho for a demo of notebooks coordinated with Prefect. https://www.youtube.com/c/PrefectIO