Hi ! i was trying to write a flow which runs a deployment (wrapped under a task) parallely for a set of input ( in array) . and i want to perform certain actions only after all these runs have completed.

from prefect import task, flow
from prefect.deployments import run_deployment


@task()
def task_A(event):
    output = run_deployment(
        name='some deployment',
        parameters={'event': event}
    )
    if output.state_name != 'Completed':
        raise Exception("flow run not completed")
    return output


@flow()
def flow_B():
    a = [1, 2, 3, 4, 5, 6, 7]
    # method 1
    for i in a:
        task_A.submit(i)

    # method 2
    output = task_A.map(a)
    for tasks in output:
        tasks.wait()

    #do something when all the parallel executions have completed for task A

this is what my flow looks like but in method 1 it runs next iteration of task_A only when previous iteration is complete in method2 it doesnt wait for the tasks to complete

Hi, with Prefect 2.0, how to use command line to run a flow with parameters (prefect deployment run id ......etc)

I'm looking to try to use a 3rd party secret store (we use Hashicorp Vault, but AWS Secrets Manager, etc would be similar). In our blocks, we want to store a pointer to the secret (path, location, etc), but not the actual value. And then retrieve the value at runtime. Has this been done before?

Hi guys! I'm running a flow using the Orion dask executor on Kubernetes but i get an error on very large datasets:

distributed.scheduler.KilledWorker: Attempted to run task prometheus_to_gcs-02d33e7b-92a6-4fe8-8258-ba9efaa3d609 on 3 different workers, but all those workers died while running it.

What could be the problem here? Other smaller datasets are running fine. I am running it with adaptive workers (min 4, max 10). I am not sure though if I used correctly to dask configuration for what i am trying to do. Can you please have a look at my code in the thread? I am able to run the same flow in Prefect 1.4. Thanks!

Hey Folks, how can I check the health of a Perfect Agent in 2.0? It is running inside a docker container and I’m trying to figure out a reliable health check locally instead of pinging the Orion server.

Does prefect 2.0 have a way to tell a task to wait for manual intervention? For workflows with humans in the loop.

deploying Azure Container Instances - I keep getting a timeout waiting on container creation. Has anyone else experienced this?

After installed Prefect Orion 2.7, I am getting the following exception and I couldn't see any of my existing flow-runs in Orion UI. TypeError: unsupported operand type(s) for -: 'DateTime' and 'NoneType'

I have a flow that is occasionally failing with the following error: State message: Given states: total=2, not_final=2, 'RUNNING'=1, 'PENDING'=1. Both of the tasks in the flow finish as completed as seen in the photo. The flow's code looks like this:

@task
def prod_9e_etl_asx_anns():
    logger = get_run_logger()
    res = invoke_lambda("prod_9e_etl_asx_anns", {}, True)
    <http://logger.info|logger.info>(res)
    return ""


@task
def prod_9e_etl_asx_ann_sentiment():
    logger = get_run_logger()
    res = invoke_lambda("prod_9e_etl_asx_ann_sentiment", {}, True)
    <http://logger.info|logger.info>(res)
    return ""


@flow(task_runner=SequentialTaskRunner(), retries=2)
def perfom_asx_anns_etl():
    x = prod_9e_etl_asx_anns.submit(return_state=True)
    y = prod_9e_etl_asx_ann_sentiment.submit(wait_for=[x])

I need to make sure the tasks happen in sequential order. Any idea how I prevent this error?

I’m trying to run

prefect agent -q tenzinga_django_prod

on my server … I’m using a

systemd

to run. It errors and lets me know that Orion (the server not agent) needs SQLite >= 3.24.0 but Amazon Linux 2 only has 3.7.17 as it’s latest version.

This message was deleted.

@Anna Geller Is there a plan to integrate celery for task distribution in roadmap?

@community, anyone used celery with Prefect for task distribution?

Dose anyone know the status of the prefect 1.0 API? I have an internal error code 500 while requesting prefect 1.0 api.

Are there any best practices when it comes to setting up a dev and production env in prefect? I was thinking to put it in two different workspaces, so it would also be easier to take advantage of blocks, but I'm not sure if i could run two agents on two different workspaces on the same pc. (Would have to login to two different workspaces simultanously?) Currently i've setup two different docker blocks that injects the secrets i need in different environment into the env vars in the container, but this would lead to exposed secrets if i was to inject secrets into the docker like this. One way would be to use different Azure vaults and just inject these into the docker, but I believe the prettier solution would be to use prefect blocks. Anyone who did something similar and can share ideas or experiences? TL:DR - What is the best way to deal with secerts/connection string and other variables that span across environments? Ex.. The Server1 Var in my Dev environment == DevServer Server1 Var in Prod var == ProdServer

Hi what is minimum hardware requirements (CPU/RAM) for Prefect production installation ? with all components (SQL/UI/Engine/Agent etc.)

Hi all, Is there a way of retrieving a flow's scheduled start time from context in Prefect2 ? As we want to use the scheduled time (not the actual run time) as input for some of its tasks. It was easy in Prefect 1 but can't work out how to do it in Prefect2 now that we've migrated. Thank you !

Thanks Anna! Sorry about that, i'll keep that in mind next time :)

Hi all, I need to run a flow which runs a stored procedure on a SQL Server 2017 instance (which is hosted on AWS) using the python package

pyodbc

However, I continue to get the error

Failed to load and execute flow run: ModuleNotFoundError("No module named 'pyodbc'")

I have.. • added

pyodbc

to requirements.txt, which is installed using our

Dockerfile

with

RUN pip install -r requirements.txt

Info: Prefect version: 1.2.4 Deployment: AWS CDK for AWS Where does your docker container live?: an ECS Fargate deployment. What OS runs on your docker container? Amazon Linux 2 Any ideas? Is Prefect just not finding the package?

Are you looking for a way to create an ML powered prediction service in minutes? In Prefect Live at 3PM Eastern today, Tim Liu, the head of product at BentoML, will talk about how to ship your models to production in minutes🚀 Come join us live on Twitch or YouTube.

We are seeing some odd behavior when connecting to Snowflake from a task using the snowflake connector. This works:

for doc in query:

, but it appears to load the entire resultset into memory before enumerating the rows. The better practice is to use

doc = query.fetchone()
while doc: 
  doc = query.fetchone()

However, this hangs immediately when run from within a task. Has anyone else seen this behavior? Any idea why it might be behaving this way?

i recently upgraded our airbyte from 40.14 to 40.23, and im now encountering an error with prefect (1.2.4) triggering airbyte runs. the error message we're seeing in our logs is below. anyone else run into a similar issue with newer releases of airbyte recently? im wondering / guessing if this is related to a change in the airbyte API.

Task 'AirbyteConnectionTask': Exception encountered during task execution!
Traceback (most recent call last):
  File "/home/ec2-user/.local/lib/python3.7/site-packages/requests/models.py", line 971, in json
    return complexjson.loads(self.text, **kwargs)
  File "/usr/lib64/python3.7/site-packages/simplejson/__init__.py", line 488, in loads
    return _default_decoder.decode(s)
  File "/usr/lib64/python3.7/site-packages/simplejson/decoder.py", line 374, in decode
    obj, end = self.raw_decode(s)
  File "/usr/lib64/python3.7/site-packages/simplejson/decoder.py", line 393, in raw_decode
    return self.scan_once(s, idx=_w(s, idx).end())
simplejson.scanner.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/ec2-user/.local/lib/python3.7/site-packages/prefect/tasks/airbyte/airbyte.py", line 67, in _check_health_status
    self.logger.debug("Health check response: %s", response.json())
  File "/home/ec2-user/.local/lib/python3.7/site-packages/requests/models.py", line 975, in json
    raise RequestsJSONDecodeError(e.msg, e.doc, e.pos)
requests.exceptions.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/ec2-user/.local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 884, in get_task_run_state
    logger=self.logger,
  File "/home/ec2-user/.local/lib/python3.7/site-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/home/ec2-user/.local/lib/python3.7/site-packages/prefect/utilities/tasks.py", line 456, in method
    return run_method(self, *args, **kwargs)
  File "/home/ec2-user/.local/lib/python3.7/site-packages/prefect/tasks/airbyte/airbyte.py", line 333, in run
    session = airbyte._establish_session()
  File "/home/ec2-user/.local/lib/python3.7/site-packages/prefect/tasks/airbyte/airbyte.py", line 60, in _establish_session
    if self._check_health_status(session):
  File "/home/ec2-user/.local/lib/python3.7/site-packages/prefect/tasks/airbyte/airbyte.py", line 76, in _check_health_status
    raise AirbyteServerNotHealthyException(e)
prefect.tasks.airbyte.airbyte.AirbyteServerNotHealthyException: Expecting value: line 1 column 1 (char 0)

So I noticed this last night as I was getting everything setup, but it would seem that my Work Queue is

unhealthy

… I went back and ensured that my starting of the agent in production was using the correct queue. It appears to be spelled correctly and everything. However, the

unhealthy

bit has never gone away and it doesn’t have a

last polled date

.

Hi there. Is there some documentation I can read on running prefect on on spot instances (GCP)? Kinda curious if that something that will work and if so, what setup is require.

I feel like the coolest kid on the block after seeing my handle in the contributors section for 2.7.0 lol

This message was deleted.

Hello, it appears that creating a flow deployment fails when I set an env var as the default value for a task input parameter. I assume that's because the environment that creates the flow deployment (GitHub Actions Runner) doesn't have that env var set. Is this a bug? I feel like I should be able to create and apply the deployment with no issues. Examples in thread.

hey all! I'd like to invite an outlook group to our prefect instance. I sent the first email and it was caught up in some email filters. I tweaked our filters but now when I resend the invite no email comes. is there any way to force send another email?

hello! In regards to using ECSTask as infrastructure are folks doing a block per flow, blocks per flow(different task definitions available to different flow deployments for a single flow) or single block with infra overrides? I want to give people the option to pull in secrets and EFS volume stuff as needed.

Hi Prefect Folks - I am running Prefect 1.x and using Kubernetes Agents with Dask pods to execute tasks. I’ve had the need to create my own

StorageSchema

. Is there a recommended approach to ensuring my custom StorageSchema is globally accessible in the

StorageSchema.type_schemas

? As a workaround we are calling

StorageSchema.type_schemas["MyCustomStorageSchema"] = MyCustomStorageSchema

in our CLI wrapper before creating agents so that the custom schema is accessible, but when we create child flows via Dask workers the custom storage is not set and the flow fails.