Hi Prefect, I want to run a flow on ECS and use GitHub as storage. My python code imports modules. The flow definition looks something like:

from tasks import my_task
from prefect.storage import GitHub
from prefect import Flow
from prefect.run_configs import ECSRun


storage = GitHub(
    repo="repo",  # name of repo
    path="path/to/myflow.py",  # location of flow file in repo
    access_token_secret="GITHUB_ACCESS_KEY",  # name of personal access token secret
)

with Flow(name="foobar",
          run_config=ECSRun(),
          storage=storage) as flow:
    my_task()

The problem seems to be that the GitHub storage only clones the single file and not the entire project which causes my import to fail. (

ModuleNotFoundError("No module named 'tasks'")

) I've seen that there has been some discussion around this issue but it hasn't really helped me to solve the issue.... Is my only option to clone the repo into the custom image that I use for my ECS task? But that would mean that I would have to rebuild that image every time I change something to my underlying modules, right?

Hi Everyone, I'm developing a handler that will be sending slack notification to our team on flow failures with detailed information. I have managed it so far and it runs flawlessly locally. The issue comes when running it on our server, the handler sends a default notification that just says

Some reference tasks failed.

I was investigating a bit and seem that the state contains the result of all tasks while running it locally (

state.result

) while on the server this is empty (i printed it using the logger) . Any idea on how to address this?

Used Map functionality, But at a given point of time, I only see hike in only one cpi. Used DaskExecutor with 8 worker and 4 thread.Can anyone tell what is the issue.

Hello - I have a Q about ad-hoc env vars when running a flow — is it possible to “save” them somehow so they can be re-used again? or do I have to refill them manually each time i wanna create an ad-hoc run?

Hi. poetry add prefect[azure] is giving error. And I am guessing this is the reason, my flows are not working. Without azure they were working fine. What can be done here?

Hello all, i had my prefect orion cloud running for last 4 days with no hiccups. today i logged in and i see all the deployments disappeared from the cloud. Is there a time limit for these processes? @Anna Geller

Hey again everyone, is anyone implementing prefect while initializing local agents inside kubernetes pods? When our kubernetes pods die off and restart (for whatever reason) the old agents continue to live in a “dead” state in prefect. Eventually the agents accumlate and slow the whole UI down and deteriorate the performance. Anyone else have this problem or know a way to manage the agents in code?

Hello couple questions, 1. After deployment, if i make further changes to my file. How does this new change get incorporated in the cloud? do i need to do a

prefect deployment create deployment_name

again for those new changes to take affect? 2) if above is true, then do i need to rerun the tasks again on the UI? 3) sometimes i inadverently press run twice and i have two running processes. Is there anyway to stop a process after it has been started? 4) when i delete the workspace, to start over, i notice when i type,

ps aux | grep python | wc -l

the python processes are still running and i have to do a

pkill python

to kill all the python processes. Is there any way that once a workspace is killed all the python processes are killed along with it?

For GitHub storage, can we use it to upload flows as we do with cloud? In the documentation, it asks for 'path' which is location of an already existing flow file. If yes, how can we do that?

Hi, I'm new to Prefect and installed a local server. Last week everything seemed to work and I started to write my first flows. They were executed as expected, also the UI worked great. However today morning, I tried to continue working on my flows, but for some reason the UI was frozen and I just restarted the containers through docker-compose (down and up again). After that I experienced a curious / weird output in the Postgres logfile (I will explain below). Even the UI was working fine again, the logfile of Postgres was quite confusing. So I thought, that something has messed up with my agent or other communication issues between Prefect server and the database. I decided to make a new clean setup in a new clean virtual environment. I also setup a path to a new database, and deleted the .prefect folder in the home directory. But even with the clean setup the weird behavior of Postgres started right away: ==> More details in thread

Hi there. I'm signed up to Prefect 2.0 cloud. I'm trying to set up a simple agent to run on a docker container. So far this is my

Dockerfile

for the agent:

FROM prefecthq/prefect:2.0b2-python3.9

RUN prefect cloud login --key cloud_api_key

ENTRYPOINT prefect agent start 'queue_id'

However, the second line requires me to specify the workspace - is there some flag I can add (like

--workspace {workspace}

) ? There probably is a much better way to set up an agent, any other docs would be appreciated! Thank you

Hi, is there a recommendation for what kind of machine specs we should use to run prefect core? I have a couple instances running on t3.mediums (in separate unconnected environments) and they seem to be failing randomly after several days

Does anyone have a documentation framework for data pipeline/testing best practices? I guess similar to dbt's style guide but for your pipeline?

Task was destroyed but it is pending! task: <Task pending name='Task-36427' ! Sometime I got this error. Note: I have millions of task running parallely.

A rundown of a variety of 😛refect: content over the next 2 weeks! Look forward to seeing you there, and don't forget to join the #events channel to stay up-to-date. https://prefect-community.slack.com/archives/C036FRC4KMW/p1649100935959699

Hi all! My company is just starting to use Prefect as our orchestration tool for our data pipeline for end customers and I have a few questions on how best to set it all up. Is there a way to reuse a single flow for every company using our product dynamically? We want to use Prefect to orchestrate each company’s data/ETL pipeline and we have the configuration for the pipeline (which system they’re integrating with etc..) stored in our database. The ideal is to dynamically create flows per company on their own schedule. Is that possible? Am I thinking about this the correct way?

In prod, what is the most typical Dask cluster set up? 1. Temporary cluster? 2. Existing cluster? If existing cluster, what is the most popular cluster manager? Yarn, K8s? Thanks!

Is there a way to provision automations as code? Ideally I’d like to check mine into source control and use CI/CD to deploy them

Observation: So architecturally, Agent sounds like the "inverse Scheduler" that lives in client's infra. Because Prefect cloud is not allowed to tunnel into client infra, Agent fulfills the role of being an extension of the cloud Scheduler that resides in client infra.

Hello! I’m new to Prefect so please excuse the newbie question. If using Prefect Cloud, do I still need to deploy an agent somewhere? (let’s say I want to trigger a dbt run on dbt cloud). From what I can see from the docs, I always need an agent running to listen for the trigger. I was wondering I could perhaps setup an API endpoint to spin up say a docker container on demand, triggered by Prefect cloud. It looks like I still need an agent running to do that, unless I’m missing the point? Thanks

Hello Everyone. Hope you all are doing well. I have just started with prefect. I have created a workflow containing consisting of 6 tasks. And they are defined as followed

def fun1(): <some code here>
def fun2(): <some code here>
def fun3(): <some code here>
def fun4(): <some code here>
def fun5(): <some code here>
def fun6(): <some code here>

def execute_flow():
	
	@task
	def t1():
		fun1()
	@task
	def t2():
		fun2()
	@task
	def t3():
		fun3()
	@task
	def t4():
		fun4()
	@task
	def t5():
		fun5()
	@task
	def t6():
		fun6()
		
	
	with Flow('my flow') as f:
		
		a = t1()
		b = t2()
		t3(a)
		t4(b)
		c = t5()
		d = t6(c)
		
	output = f.run()
	
	result = output.result[d]._result.value
	return result

the expected behaviour of flow would be t1 -> t2 -> t3 -> t4 -> t5 -> t6 but it is not working this way instead this is what happenning t1 -> t5{fails but shows 'success'} -> t6{fails since result from t5 are not appropiate} -> t2 -> t3{shows success message without even going through half way of function} -> t4{same result as t3} except t5 and t6, every task is time taking

Hey folks, I have a false positive error occurring when a flow runs. It concerns an SQLAlchemy Result object not being pickable. The flow achieves it’s objective successfully, with data inserted into a database but I still get this error. What is Prefect pickling at runtime that could be the cause of this error. Here’s the specific output from the logs:

Unexpected error: TypeError("can't pickle sqlalchemy.cprocessors.UnicodeResultProcessor objects")

Hi all 👋 Just wondering: are 3rd party tasks (i.e. the Great Expectations validation) compatible with Prefect Orion? Anyone using this? I’d want to validate a dataframe from a upstream task with GE in a flow… Cheers, C

Hi, I’m trying to follow this blog post here and I’m failing at the first hurdle. When I run

prefect auth login --key <replaced with my key>

I get the following error:

Unauthorized. Invalid Prefect Cloud API key.

The cloud UI says my key has been created

Hi all, I've a question regarding running a Flow: I've set up Prefect-Server at Rancher, the workloads seem to be stable, UI can be reached, an Agent is registered - everything should be fine. Now I'd like to run a simple Flow on the Prefect-Cluster, using the Python script and using the Prefect Run command - unfortunately I get:

└── 14:27:40 | INFO    | Entered state <Failed>: Failed to load and execute Flow's environment: ModuleNotFoundError("No module named '/home/prefect-flow-example'")
Flow run failed!

The file "prefect-flow-example.py" is available at my local host, I've set the proper ENV variables for the connection to the Prefect-Server. Do I need to define some Storage Options before? An Azure subscription would be available. I'm curious about what you would suggest as best practice 🙂 Thank you a lot in advance! Patrick

Hello. Is there a Python API to get the flow version given a flow name? Or is flow version only accessible via the

prefect get flows

cli command?

Greetings, is anyone else having issues logging in to prefect.io? Using google auth, i've tried multiple browsers, cleared browser cache. Seems to be spinning

Hi, I’m currently pulling a dbt repo with pygit2, then executing dbt using the DbtShellTask. We have a profiles.yml in the dbt repo, but I wanted to ask if there was a way to dynamically make a profiles.yml. The idea is to have a prefect flow where users can input in a schema name (normally defined under

schema

within the profiles.yml), and have the dbt task within that flow output to a schema of the user’s choice.

Hi, I’m following the instructions in this great article on using ECS. I’m now setting up a custom docker image to run dbt (I’m assuming I can’t use

DbtShellTask

because dbt wouldn’t have been installed). It’s coming back with an error where I need to specify a

execution_role_arn

in the

ECSRun

which I’ve now done, but can I use the previously defined role, or do I need to define something else? Thanks!