prefect-community #prefect-community

Alex F

05/17/2023, 11:29 PM

Copy code

Hello All , Anyone see this error starting prefect server v1 ? ERROR: for hasura  Cannot start service hasura: exec: "docker-init": executable file not found in $PATH
 prefect server start
Pulling postgres ... done
Pulling hasura   ... done
Pulling graphql  ... done
Pulling apollo   ... done
Pulling towel    ... done
Pulling ui       ... done
Creating network "prefect-server" with the default driver
Creating tmp_postgres_1 ... done
Creating tmp_hasura_1   ... error

ERROR: for tmp_hasura_1  Cannot start service hasura: exec: "docker-init": executable file not found in $P                                         ATH

ERROR: for hasura  Cannot start service hasura: exec: "docker-init": executable file not found in $PATH
ERROR: Encountered errors while bringing up the project.

Shivam Tiwari

05/18/2023, 2:18 AM

Copy code

Hi All,
Facing an issue with prefect mapping of inputs to task.
save_report = save_to_db.map(
         combined_reports_list , upstream_tasks = [add_report_signal])
In this the length of combined_reports_list is 5 and so the task save_to_db is expected to run 5 times but it runs only once, any possible reason for this ? Also when i remove the upstream_tasks it works as expected.
I am using prefect version 0.15.13.
Any help would be appreciated on this. Thanks

Karthik

05/18/2023, 9:00 AM

Hi everyone, I’m trying to run subflows in parallel. These subflows in turn call other flows. This is somewhat of a setup I have,

Copy code

import asyncio
from prefect import flow

@flow
async def subflow_1():
    print("Subflow 1 started!")
    some_other_flow()
    # post result ops

@flow
async def subflow_2():
    print("Subflow 2 started!")
    some_other_flow_2()
    # post result ops


@flow
async def subflow_3():
    print("Subflow 3 started!")
    some_other_flow_3()
    # post result ops


@flow
async def subflow_4():
    print("Subflow 4 started!")
   some_other_flow_4()
    # post result ops


@flow
async def main_flow():
    parallel_subflows = [subflow_1(), subflow_2(), subflow_3(), subflow_4()]
    await asyncio.gather(*parallel_subflows)

if __name__ == "__main__":
    main_flow_state = asyncio.run(main_flow())

Now when I deploy this (or run locally), flow runs are created for the 4 subflows, but that’s about it, thereafter it’s all sequential, nothing gets processed parallely. For Eg: subflow_2 run proceeds only after subflow_3(in turn some_other_flow_3) have finished. So in the end it makes no difference to calling each of these subflows sequentially instead of using asycio Am I missing something else here? Kindly advice

Gregory Hunt

05/18/2023, 12:25 PM

Where does the code the

if __name__ == "__main__":

block on code run. Does it run on the Agent? I am working on setting up a streaming flow with GCP Cloudrun jobs for the worker pool and am curious sync this part of the flow is continuous?

Chris Gunderson

05/18/2023, 2:14 PM

We are starting to migrate to Prefect 2. I was setting up the directory structure to have blocks, deployment scripts, flows, tasks, and utilities. The flows folder will have a sub-folder to break up the projects. The same docker image will be used for all project, mainly due to the overlap of libraries we use.

Slackbot

05/18/2023, 3:00 PM

This message was deleted.

Ofir

05/18/2023, 5:07 PM

What’s the best practice to provide input files to a Prefect run? Should I simply upload to e.g. S3 / Azure Files and then pass a URI to the flow (a workflow parameter)? Or perhaps the

prefect

CLI has support for doing both in the same command?

✅ 1

David

05/18/2023, 5:46 PM

It appears that a 2.10.9 agent does not work with a 2.10.8 server (at least in oss). I was running a local server with local agent and although the agent said it connected, the queue was shown as unhealthy and no work was pulled. When I downgraded the agent to 2.10.8 everything started working. Are there local agent logs I could have looked at to figure this out ? I could not see logs on the server or agent that indicated an issue.

kasteph

05/18/2023, 5:50 PM

Has anyone been seeing connection issues with Prefect 2.10.7?

Copy code

Backing off due to consecutive errors, using increased interval of  60.0s.
ConnectionState.CLOSED
	
httpx.LocalProtocolError: Invalid input ConnectionInputs.SEND_HEADERS in state 
    raise mapped_exc(message) from exc	
line 77, in map_httpcore_exceptions
  File "/usr/local/lib/python3.9/site-packages/httpx/_transports/default.py"

✅ 1

05/18/2023, 7:13 PM

Question: now sure where the is should go If I have 3 different flows. How can I pass some kind of correlation? I’ve got an api which kick starts a flow. Then once the api has finished, it’ll report back to prefect to schedule another separate flow. Now I want to use prefect to look across the 2 flows and see what happened. Eg, if one failed. Where did it fail. Was flow 1 or flow 2? Can I tie them together in this way?

✅ 1

Choenden Kyirong

05/18/2023, 8:41 PM

hey folks! I currently have the slack webhooks integrated which is great for notifying statuses about flow runs/states. I’m wondering if there is a way to send a request elsewhere when the flow is running/stopped/completed…etc? For example, is it possible to ping a server that the flow has completed?

✅ 1

David

05/18/2023, 9:34 PM

In the environment I work in we are required to assume roles and get temporary credentials to access AWS resources: https://www.learnaws.org/2022/09/30/aws-boto3-assume-role/ Am wondering if it is possible to write a new block that takes just the IAM role that it should assume, and then allows the agent using the block to assume that role, get AWS creds, connect and pass the object back to the flow, and then the flow can perform whatever operation it needs . I don’t know much about how blocks interact with agents, but does that sound feasible ?

Ajeel Ahmed

05/18/2023, 10:08 PM

Hello again! Yes I’m back 😅 The problem: Pointing to the private registry (a

DockerRegistry

block) through the

image_registry

parameter of

DockerContainer

doesn’t work, I can tell that by looking at the Docker registry field in the GUI is still empty. This means I have to manually go to the GUI and point it to the registry. I don’t want to set it myself each time, how do I accomplish this through the library?

✅ 1

Pedro Machado

05/18/2023, 11:32 PM

Hi there. I am having an issue with a local (process) worker crashing with this error:

RuntimeError: Service exceeded error threshold.

It's crashing while running a dbt job that takes about 30 min. The worker is running on my desktop computer (Ubuntu running on WSL 2). This is what I am running.

Copy code

Version:             2.10.10
API version:         0.8.4
Python version:      3.10.8
Git commit:          8159450b
Built:               Thu, May 18, 2023 3:43 PM
OS/Arch:             linux/x86_64
Profile:             cloud
Server type:         cloud

Any ideas about what may be happening?

✅ 1

Deceivious

05/19/2023, 12:37 PM

Hello, 1. Does prefect server have services to clear unused database elements? [e.g. expired cache and unused block documents] 2. Does clearing deployments clear its infrastructure blocks? 3. ^ Similar question with task runs and cache blocks in db?

Ajeel Ahmed

05/19/2023, 2:19 PM

hey guys still waiting for a feedback, you can find the problem here

Deceivious

05/19/2023, 2:35 PM

Cross posting for community opinion as well

Ryan Brennan

05/19/2023, 3:20 PM

It looks like all of our 1.0 Starter accounts had their schedules turned off on May 15th..

Robert Banick

05/19/2023, 3:50 PM

Does anyone have experience with DevOps operations within the context of a flow? My team uses Github branches of our main ETL repo to develop new features or fixes for our ETLs. In our previous orchestration stack we could swap these branches in to the container server we spun up for a given ETL run. This helped us test the feature/fix in a production environment before merging to the ETL repo

main

. Now setting up Prefect I’m having trouble designing a similar system. I’m using Prefect 2 on AWS Elastic Container Service (Fargate) tasks to implement ETL runs. We install our ETL repo as a library onto a docker image that the

ECSTask

block Task Definition uses. I’ve tried replicating our previous system by running

pip install --upgrade git@<repo>@<branch>

to upgrade the package in question at the very beginning of my flow. This works and I’m even able to see that the function I’m modifying is indeed updated in

/usr/local/lib/python3.10/dist-packages/<package>

. Nevertheless, when my flow run reaches the crucial step I’m testing, it very clearly uses (and fails on) the “old” function currently in the ETL repo

main

. Forcing reloading the repository in question via

importlib.reload(<package>)

does not appear to resolve the problem. My questions therefore are: 1. Is it possible to change a library mid-flow like this or is it a hard limitation of Python / Prefect? 2. Is Python code used by flows somehow installed somewhere different from

/usr/local/lib/python3.10/dist-packages/

on the container? Such that

pip

would install in the wrong place… 3. If it’s not possible to change a library mid-flow, is it possible to have the Prefect Agent/Worker run the

pip

installs prior to spinning up the flow? Could the Agent even read the desired branch names from the flow parameters? 4. Any other ideas? The nuclear option here is manually changing the branches on docker images but that’s very clunky and will make iterative testing extremely time consuming. So we’d really like to avoid that path. All help and suggestions most appreciated, Robert

kiran

05/19/2023, 5:26 PM

Hi y'all. I have a script running that seems to be functioning alright — the logs in Cloud don't any issues. But my piped output shows

16:50:25.956 | ERROR   | prefect._internal.concurrency.services - Service 'EventsWorker' failed.

Any idea what's going on?

✅ 1

Jenia Varavva

05/19/2023, 6:03 PM

Hi, I'm running an agent with arguments like

prefect agent start --pool data-qa

and some env vars (as set by prefect-agent helm). The agent picks up work from the queue and launches it fine, but the cancellations don't work. Looking at the agent code, there's a chain of calls: check_for_cancelled_flow_runs -> Agent.get_work_queues() -> update_matched_agent_work_queues() with the last one containing a condition on the whole method body:

Copy code

if self.work_queue_prefix:

and not doing anything otherwise. Since I don't set work_queue_prefix on the CLI, the agent ends up not matching any queues, so the flows in

Cancelling

state never get processed. Is this a bug or am I doing something wrong?

Dominic Pham

05/19/2023, 7:00 PM

Hi all, is there a way to remember a value between scheduled flow runs? Let's say I am iterating through a list of states and want to check off the states that have already been iterated between flow runs so that we don't rerun the flow for the same state twice.

Dominic Pham

05/19/2023, 7:01 PM

Ideally these would be scheduled in sequence rather than run in parallel

sjammula

05/19/2023, 7:35 PM

@prefect Team, can someone provide me info regarding

skip_if_running

how it is handled in prefect2. I'm looking for workaround for skipping the deployment .I did notice we have concurrency limit set at work pool/queue level but I require at deployment level.

Anders Segerberg

05/19/2023, 8:19 PM

Prefect 1.0 With a LocalExecutor, will a

.map()

run over its arguments in-order? Is there a guarantee of that? Or is there no guarantee on the iteration order

Faheem Khan

05/21/2023, 12:41 AM

perfect 2.10.10 I have the results showing a blank page.

Oliver E

05/21/2023, 3:04 AM

Does anyone know how i can delete my prefect.io account. I do not see an option to do so

Alessandro De Rose

05/21/2023, 10:22 AM

Hello,

mypy

is complaining when attempting to unpack a tuple returned from a task. In the example below, when attempting to unpack

from the call to

get_numbers

mypy

raises the error

"None" object is not iterable

. However, the type definition for the

get_numbers

function shows that

None

should never be returned from the task. Any idea what's wrong here? (Prefect v2.10.10, mypy v1.3.0)

Copy code

from prefect import flow, task


@task
def get_numbers(a: int, b: int) -> tuple[int, int]:
    return a, b


@flow
def get_numbers_flow() -> None:
    a, b = get_numbers(1, 2)   # <- "None" object is not iterable.

Nitin Bansal

05/22/2023, 2:56 AM

Hi All, I am in process of moving from Prefect 1 to Prefect 2. I came to know that Prefect 2 dashboard has a new url https://app.prefect.cloud/ as compared to Prefect 1 https://cloud.prefect.io/. Is there any way I can migrate organizations and all the settings including payment etc from Prefect 1 to Prefect 2?

✅ 1

Nitin Bansal

05/22/2023, 6:51 AM

ImportError: cannot import name 'apply_map' from 'prefect', getting this error while upgrading to Prefect 2.0. Try to find out but not able to get substitute of apply_map in Prefect 2. Need Help