Hi, I am making my first pass at using Github actions to CI. I have reached something I don’t fully understand. On pull_request of a feature/** branch, I run actions that: 1. checkout the branch 2. create a project in prefect cloud 3. register the flow.py with a --label dev when I run the flow manually, i get an error:

Unexpected error while running flow: KeyError('Task slug connect_source-1 is not found in the current Flow. This is usually caused by a mismatch between the flow version stored in the Prefect backend and the flow that was loaded from storage.\n- Did you change the flow without re-registering it?\n- Did you register the flow without updating it in your storage location (if applicable)?')

See you all in 5 mins on a Fireside Chat:

Hi, do you guys know if it’s possible to pass

upstream_tasks

with methods that actually requires parameter? The documentation is not really clear regarding how to chain multiple upstream_tasks to create dependencies between them and I was wondering if some of you had some more complex examples. For example, I don’t think I can do this

with Flow(f"{environment_prefix}-test", storage=S3(bucket=storage_location_bucket_name)) as flow:


    dbt_run(organization_id_param, data_processing_start_date_param, data_processing_end_date_param, should_process_last_period, period, period_value,
            upstream_tasks=[pull_snowflake_secret(a,b), pull_repo(b,c)])

Does anyone have a complete working example on how to use PrefectResult? Its simple: Scrape a website for a link, if the link is different than the prefect result, download it and save the new link as the result. Otherwise exit with "success: no work to be done". I don't want this to be a local result because we plan to move these flows off of a windows agent and into docker agent. Thanks in advance!

is there an equivalent of

IMAGE_PULL_SECRETS

for the Orion k8s agent/flow runner? I couldn’t find anything in the docs and the code

Hello, for prefect 2.0 - if I want to add flow run id in all the log statements, do I have to override format of all the loggers defined in default logging config or is there a way to apply this change by overriding only one parent logger?

Hi all! I am new developer and an new to prefect. Currently we are trying to implement prefect for our ETLS at my job. We use vmware and windows sever, we found out the hard way that using prefect with nested virtualization does not work well. So now are moving to prefect cloud, using local agents. Will this work? Is it necessary to run multiple agents for multiple flows or can one agent run multiple flows?

Hi, in prefect 0.14.11, prefect execute flow-run downloads the necessary local dependencies from the repo when using GitLab storage but this behavior seems to have changed in at least 1.0.0…could you help remind me of when this was changed? Also does this mean that we have to build a docker image for each repo we have? :(

does configuring a flow with

KubernetesRun

restrict the flow only to kubernetes agents or can it still be picked up by any agent as long as the labels match? I was under this impression after reading the docs but when I created 1 docker agent and 1 k8s agent with the same set of labels, my kubernetes flow got picked up by the docker agent

Has anyone used Great Expectations, how does it compare to other testing frameworks?

Hi all! I followed the "Running flows in Docker" tutorial for 2.0 (

prefect==2.0b2

, Python 3.9, Linux): • created an S3 bucket and a storage (isn't it too much for a local run?) • started a local API server and configured PREFECT_API_URL • created a queue • started an agent for that queue • created a deployment with DockerFlowRunner and no parameters Then I ran the deployment and the agent output wasn't very informative, see below. In the UI the flow run "dag-sample/caped-antelope" is shown as failed without any logs available. Anyone can explain?

$ prefect agent start 6aa0c2b2-e895-4b8c-aed3-8bef5b2c88ab
Starting agent connected to <http://127.0.0.1:4200/api>...

  ___ ___ ___ ___ ___ ___ _____     _   ___ ___ _  _ _____
 | _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
 |  _/   / _|| _|| _| (__  | |    / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|   /_/ \_\___|___|_|\_| |_|


Agent started! Looking for work from queue '6aa0c2b2-e895-4b8c-aed3-8bef5b2c88ab'...
13:29:31.446 | INFO    | prefect.agent - Submitting flow run '8eb8aa0b-82da-4081-9a42-aae1f22b0525'
/home/tarkovskyi/miniconda3/envs/prefect_exp39/lib/python3.9/site-packages/prefect/flow_runners.py:697: UserWarning: `host.docker.internal` could not be automatically resolved to your local ip address. This feature is not supported on Docker Engine v19.3.15, upgrade to v20.10.0+ if you encounter issues.
  warnings.warn(
13:29:32.215 | INFO    | prefect.flow_runner.docker - Flow run 'caped-antelope' has container settings = {'image': 'prefecthq/prefect:2.0b2-python3.9', 'network': None, 'network_mode': 'host', 'command': ['python', '-m', 'prefect.engine', '8eb8aa0b-82da-4081-9a42-aae1f22b0525'], 'environment': {'PREFECT_API_URL': '<http://127.0.0.1:4200/api>'}, 'auto_remove': False, 'labels': {'io.prefect.flow-run-id': '8eb8aa0b-82da-4081-9a42-aae1f22b0525'}, 'extra_hosts': {}, 'name': 'caped-antelope', 'volumes': []}
13:29:33.547 | INFO    | prefect.agent - Completed submission of flow run '8eb8aa0b-82da-4081-9a42-aae1f22b0525'
13:29:33.584 | INFO    | prefect.flow_runner.docker - Flow run container 'caped-antelope' has status 'running'
13:29:44.923 | INFO    | prefect.flow_runner.docker - Flow run container 'caped-antelope' has status 'exited'

I'm struggling to implement the GCP_CREDENTIALS default secret feature in a local Prefect flow. • I've set an environmental variable PREFECT__CONTEXT__SECRETS_GCP_CREDENTIALS pointing to a full local path for the GCP credentials .json downloaded from the GCP project to which I'd like to upload some files. • I've run the following:

from prefect import Flow
from prefect.tasks.gcp.storage import GCSUpload

with Flow("google-cloud-test") as flow:

    GCSUpload(bucket="test-upload", create_bucket=True)(
        data="test-file.csv", credentials="GCP_CREDENTIALS"
    )

flow.run()

and received an error:

AttributeError: 'str' object has no attribute 'keys'

. I've incorporated the feedback from a similar error discussion recently, but can't figure out what the issue is. Any help is much appreciated.

Hi Everyone, we have a prefect server running on Kubernetes, setup was done using the HELM Chart. Everything is running smoothly but occasionally we would get this error

File "/usr/local/lib/python3.7/site-packages/urllib3/connectionpool.py", line 341, in _raise_timeout
    self, url, "Read timed out. (read timeout=%s)" % timeout_value
urllib3.exceptions.ReadTimeoutError: HTTPConnectionPool(host='prefect-apollo.prefect', port=4200): Read timed out. (read timeout=15)

This happens especially when we use the Task StartFlowRun it does not happen very often , but I was wondering if there was a way to force retry or if anyone knows why this would be happening? Thanks

Is there a way to get logs from code running within dask? I know that for "general" use of dask, the logs would have to be pulled from the execution environment such as k8s (as noted here) When I am running code that uses

worker_client

and then submit new tasks to the same cluster (code inside), I see that printout from the first call of the function is logged in prefect and shipped to server, but consecutive calls are not (see log inside). Is there some trick I can use?

Is there an example template of what a properly architected git repo looks like for a Prefect project? I am having trouble conceptualizing the development workflow. I am imagining a "production" branch that at any given moment contains a portfolio of all the relevant flows for a given project. Changes (i.e. merges) to the production branch would trigger incremental registrations of the modified flows driven by some sort of backend "build" job triggered by the successful merge. Other things I am considering would be a way to have proposed changes to production flows go through automated testing before being merged to production and ideally a way for developers to test their flows in development without sullying the production environment. Bonus points for being able to code the tasks and flows in such a way so that they do not require editing between environments. is there a recognized reference architecture for such a build strategy?

Hi, I want to de-register some flows programatically based on certain conditions. Is it possible to: 1. Get a list of all registered flows in python (i.e. the Python equivalent of

prefect get flows

), and 2. De-register some of these flows through python.

Hi - is there an explanation of what exactly Workspaces in Prefect Cloud 2.0 do/are for? The docs I could find say how to create one but not really why you’d want to.

hey everyone, looking to change our ETL flow from celery alone to prefect, and i came across this discussion on github: https://github.com/PrefectHQ/prefect/issues/1689 we are looking to have the easiest transformation into prefect and would like to have minimum changes, but i could not find a way to use a CeleryExecutor like Dask has. Am i missing something? is it supported or should i implement it on my own? Thank you 🙂

Hi! What is the equivalent of using a json prefect secret in the cloud, if I want to do this in the local secret (.prefect/config.toml)?

Hi! How do I mount AWS credentials to

KubernetesFlowRunner

?

Lots of good fixes in 2.0b2

https://orion-docs.prefect.io/concepts/tasks/#configuring-concurrency-limits On this page it says to use prefect concurrency_limit; however, the actual command is concurrency-limit

Also, is it possible to limit concurrency at the deployment level? I am implementing a job that I don't want running twice at the same time. It seems tags in a DeploymentSpec don't work for concurrency limits. Right now I might move the tag to the main flow

This may be (or rather, is) an entirely “opinion” question: I often have the situation where I need a modern substitution for a Makefile. Don’t get me wrong, the old-school GNU Make is great at what it does, but the syntax can be, ahem, the polite word is “painful”. I would be curious to gathering some ideas: Prefect has something akin to dependency management, but there are other Python-based tools for this as well: Invoke, for example. Note that these are strictly limited to problems that will never need any sort of massive orchestration, so I would be curious what the community thinks: can Prefect solve the Makefile problem as well, or would that be bending a tool to do a task it is not designed for? And, have a nice start to the week!

Hi! I'm trying to implement a python library to gather all the tasks in common in my ETL flows and looking for some insights on the best way to do it with prefect. Sharing more in thread :

What’s up prefects - I have a pretty simple task

@prefect.task()
def get_backfill_or_scheduled_date():
    time = prefect.context.get("backfill_time") or prefect.context.get("scheduled_start_time")
    print(time)
    print(type(time))

However I see that this task creates a pod in kubernetes with the image

prefecthq/prefect:0.15.12

, is there a way that I can specify the python version for this image? The mismatch between the build and run environment is causing an error

Hey, considering the new "Beta" stage of Orion, do you still recommend implementing 1.0 in production or you think the Beta version of Orion is viable ? We are currently building our data stack and we are thinking about Prefect.

Hello all, in our development environment currently we are using a ECS Agent and have a flow of flows, that is running a large-ish batch of small pipelines concurrently, however we are getting a mix of Throttle Errors (RegisterTaskDefinition, DeregisterTaskDefinition). My agent has the AWS_RETRY_MODE and AWS_MAX_ATTEMPTS env variables as 'adaptive' and '25' respectively, but I'm still getting this error, I'm up for splitting this up into smaller calls, but I'm curious if this is a situation anyone has come across and how it was solved. How did you implement the wait/backoff logic on the flow, and is there a way to iterate with waits to implement less calls to the aws api?

Hey guys, i have 2 parameters that at least one should be provided. Prefect have a built-in function to handle this case or should i create a task and validate inside? Thanks!

I think a have a challenge. I made a significant change to an existing pipeline on the 18th which was last Friday, on this day after the change when I looked at my prefect cloud dashboard I observed it now has a new version. Fast forward to today, according to my schedule the flow ran today but the csv file imprint Friday’s date I.e

<tel:18-03-2022|18-03-2022>.csv

instead of

<tel:21-03-2022|21-03-2022>.csv

like I have said before my code is on ec2 instance that through a lambda start and stops at certain time of each day. What could be amiss?