Hey, So I have been working with custom blocks which I have now created a python package with a

pyproject.toml

for and installed through poetry into the other projects I want to use them in. It all seems to be working except I get a warning:

UserWarning: Block document has schema checksum sha256:3867078ba2512b836bbfbc1ee1831b44fe3e1dc4ea93da04e96ed0be9e1ef1e5 which does not match the schema checksum for class 'GCPServiceAccount'. This indicates the schema has changed and this block may not load.
  return cls._from_block_document(block_document)

Should I worry about this?

Hey. I'm looking into how to create a

KubernetesJob

block and apply it to my deployment through Prefect's python API. When creating the block through the Prefect Cloud UI I can get it to work fine (pull image from ECR and run it on a EKS pod). However, when creating through the Python API my flow fails to start, without any logging. Any ideas? This is what I am doing

project = "prefect20"

k8s_job_block=KubernetesJob(
    image="<AWS-ACCOUNT>.<http://dkr.ecr.eu-west-1.amazonaws.com/kubernetes-job-block:latest|dkr.ecr.eu-west-1.amazonaws.com/kubernetes-job-block:latest>"
)
k8s_job_block.save(project, overwrite=True)

# Create Flow deployment
s3_block_loaded = S3.load(project)
k8s_job_block_loaded = KubernetesJob.load(project)

deployment = Deployment.build_from_flow(
    flow=log_flow,
    name=project,
    storage=s3_block_loaded,
    infrastructure=k8s_job_block_loaded,
    work_queue_name="dev_queue"
)
deployment.apply()

In my mind it must be credential related but unsure how/where to set them up. If I go to the UI and edit the block (any setting), my flow works. Something must be done under the hood in the UI.

Another one. In Prefect 1.0 I was able to build a Docker image and push it to my registry. E.g

Docker(
    registry_url=registry_url,
    image_name=image_name,
    image_tag=image_tag,
    files=workflow_files,
    env_vars=env_vars,
    extra_dockerfile_commands=pip_commands
).build(push=True)

I understand the new concept on 2.0 where we reference a block, which references an image. But by any chance are there utilities to help building and pushing workflow base images?

Hi Prefects, One of my flows crashed with an

httpx.WriteError

. After running for about 1h and completing 371 tasks the flow entered a crashed state with the following trace:

Hi I try to start Orion on my PC (windows 11) with docker and its running but looks as is missing some settings

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

how to check the IP address what I need to use for Orion ? I was thinking that that more related with docker not with Prefect but maybe some one will help me with this :)

What would be the minimum hardware requirements for a Prefect agent?

I just tried deploy the flow to prefect cloud. I found one different is that during local orion run, I can execute sync task and async task within a flow. When I deploy, the flow will stop at calling the sync call and I must need to use async task.

hi! i tried the latest default image for prefect on our self hosted instance running on aws eks. im not able to figure out why it is trying to access resources in some another namespace ("kube-system") while we have correctly mentioned the namespace in agents as well as the job manifest. attaching the error log in thread

Hello Prefects, I’am switching from Prefect 1 to 2 one. And I’m blocked with the DAG free conception. One part of my flow is predefined rigid DAG. How can I achieve “dag behavior” in Prefect 2? Toy example of my graph:

graph = {
  "feature_1" : [],
  "feature_2" : [],
  "feature_3" : ["feature_1", "feature_2"],
  "feature_4" : ["feature_3"],
  "feature_5" : ["feature_4"],
}

So how can I “explain” that in the first time we can run in parallel

feature_1

and

feature_2

, but not the others one. We can run

feature_3

, only when

feature_1

and

feature_2

was successfully executed, and feature_3 take like input parameters the results of

feature_1

and

feature_2

. P.S. In production it should be run with

DaskTaskRunner

. Could be there any difficulty?

Hi 👋 . I’m having trouble with Prefect Agents not reporting failed KubernetesJobs back to Prefect Cloud. I’m specifically trying to ensure pods do not consume too much memory, so have the following test. expected behaviour: • Given a job with pod configuration to limit memory to 1GB. • When I run a flow which creates 2GB of memory • The flow run fails. actual behaviour: • Kubernetes correctly kills the pod, with OOMKilled status. • The agent detects the job failed. • But it does not report it as failed to Prefect Cloud, it stays as ‘running’ forever. agent log snippet:

11:32:09.162 | INFO    | prefect.infrastructure.kubernetes-job - Job 'prefect-standard-kubernetes-jobdmmq2': Pod has status 'Pending'.
11:32:11.003 | INFO    | prefect.infrastructure.kubernetes-job - Job 'prefect-standard-kubernetes-jobdmmq2': Pod has status 'Running'.
11:32:21.064 | ERROR   | prefect.infrastructure.kubernetes-job - Job 'prefect-standard-kubernetes-jobdmmq2': Job did not complete.

pod snippet:

prefect-standard-kubernetes-jobdmmq2-cpfz4           0/1     OOMKilled   0          23m

Is there some configuration I need to do to here, or is this unsupported, or a bug?

Just wanted to share kudos on the implementation of the integration (2.0) plugins. The light touch integrations, hooks to the original connectors has been great… Example, when we use the snowflake connector, the thought placed to pass through parameters to the original connector and us having access to the original connector is super useful. I’m not sure if everyone would appreciate this at the outset but, some day when you’re using these and you get access to the root connector parameters and APIs you’ll smile to yourself.

Currently when I deploy multiple flows I basically loop over them and for each, build and apply the flow. Also found it on one of @Anna Geller’s posts/examples (https://github.com/anna-geller/dataflow-ops/blob/main/.github/workflows/main.yaml#L35). [I’ve put my version in the thread] This works fine but I was wondering if this is the preferred method? My concern is that I basically duplicate the storage-block (codebase) for each flow on the bucket. Is it possible to have a shared storage-block? It seems waste of deployment time to build the storage repeatedly. I understand that this course makes sense if you have different versions or not deploy all flows at the same time.

How do I get each deployment to store its files inside its own directory using S3 block storage?

I'm having a problem with a task run using an old version of a deployment. We deployed a flow and run it. When we updated the deployment and applied it, its still running the old version. I've looked at the main-deployment.yaml and compared it to the version on S3 and its uploaded correctly. I've compared the version hash in the file to the one shown in the deployment in the UI. The flow id from the run matches the flow id of the deployment and checking the agent log the run id there matches the run id shown in the ui. I've even restarted the agent. Any ideas?

Hello, I am having.a strange issue with building a prefect deployment via a github action. The name of the flow entrypoint is complete_s3_stitch_job() but there is an error saying it can't be found in the script. However, flows: 'complete-s3-stitch-job' can be found. These are the screenshots showing the flow in the script and the error.

any ideas how to solve that ?

RuntimeError: Cannot create flow run. Failed to reach API at <http://127.0.0.1:4200/api/>.

So I’m getting some interesting “warnings” … I don’t know why. I don’t have any flows with conflicting names. So I don’t know why this is popping up. At least I don’t think I do… maybe this is left over from flows that were in testing on my local box versus on the production Prefect server.

I'm getting a weird error even though my flow exists

prefect.exceptions.MissingFlowError: Flow 'tool' not found in script 'flows/tool.py'. Found the following flows: 'monolithic-tool-flow'. Check to make sure that your flow function is decorated with `@flow`.

I have the flow defined as

flows/tool.py
-------
@flow(name="monolithic-tool-flow")
def process_tool()...

and deploy like

prefect deployment build flows/tool.py:process_tool -n process_tool  -q test -ib docker-container/local --override image=sha256:4e3346168cs3a3c6c0fb256d6e705c6629a41f0f6f23405c33a0bbc3deb964690   --apply

Hello Team so I am facing this issue where my flows were deployed and I stopped the agent for a couple of days and it didn't work as I didn't wanted them to but now that I have redeployed it I did delete all of the previous instances, yet what's happening is I am getting double entries everytime my flow runs, anyone faced this issue?

anyone getting this warning while running flows on AWS EKS ?

/usr/local/lib/python3.9/runpy.py:127: RuntimeWarning: 'prefect.engine' found in sys.modules after import of package 'prefect', but prior to execution of 'prefect.engine'; this may result in unpredictable behaviour
  warn(RuntimeWarning(msg))

hey prefect we are testing v2 with kubernetes and the agent appears to be trying to list namespaces in kube-system we are using 2.7.1

any reason it requires to look at kube-system?

I have another issue that may not be related to Prefect, but thought I would post here in the off chance someone has had this issue when using Prefect. I have a task that looks at about 30-50,000 records total if it looks at all of the records in the table. Transactions are to be committed in blocks of 1,000 records checked. For each 1k there are less than that being updated. Previously at times this task would take a few hours when it really had to go over 3-5,000 records. The first three days I started this task it took a while which was to be expected as I stopped manually running the task. However, it is now again taking a LONG time and I’m ending up with MySQL Lock Table Timeouts

MySQLdb.OperationalError: (1205, 'Lock wait timeout exceeded; try restarting transaction')

… any thoughts or ideas on what might be causing additional delay?

I'm trying to set

persist_result

to False since the object it return is not compatible with the Pickle serializer. However, I'm receiving the following error. I'm running on 2.7.2.

Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "<frozen importlib._bootstrap_external>", line 883, in exec_module
  File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
  File "/opt/prefect/main.py", line 6, in <module>
    from steps.parse.parser import parse
  File "/opt/prefect/steps/parse/parser.py", line 9, in <module>
    from .html_parser import process_html
  File "/opt/prefect/steps/parse/html_parser.py", line 22, in <module>
    @task(persist_result=False)
TypeError: task() got an unexpected keyword argument 'persist_result'

Hi all, these might be a few questions that get asked a lot but I am just starting to contribute to open-source projects so please help me out. Are there any best practices or conventions I need to follow if I want to contribute to this project? A few questions that I have are: What is the branch naming convention? Is there any commit message convention like adding issue-id: commit-msg? From which branch should create a new branch for the issue I will be working on and which branch should I raise a PR against? I understand that the main branch is unstable so probably not the best idea to branch from there. Feel free to add any other important conventions that I need to be aware of.

Hi all -- I'm self-hosting prefect via kubernetes (and using aws RDS for db) and I realized my db has been steadily running out of space since we have a lot of regularly running flows. Is there something I can set like a retention period so that flow run data from older than x days is deleted automatically?

hi all. i have one question for people running prefect orion on their self hosted instances.. do you guys get a lot of 502 errors? any possible way to avoid these ? adding retries in tasks and flow does help but it would be great if they could be avoided all together. thank you

Suggestion to add

.git

to default .prefectignore

Does anyone have a good example of using trio/asyncio directly in a task/flow? I have a basic 4-function script that uses trio memory channels. Specifically, it uses a producer/consumer with send/receive channels to achieve 40-fold concurrency for API scraping. When I try various approaches to convert to flows/tasks either hangs after completion or complains about lacking an event loop. I've tried a number of combinations and haven't found anything, including just a flow function that calls the other functions.

Hi, I am calling a function twice in a sequential flow, each call is using different input parameters. Seems like first sets of input parameters are cache and are used for 2nd calling of the function. How can I disable caching of input parameters?