@Anna Geller I have a flow with three tasks the first two ran successfully-looking at it from the UI, however the third and last task which essentially writes all the data to a parquet file in s3 is still in running state. However 23 out of 104 delivered to s3. what could be wrong expecially as cloudwatch log “delivering data to s3”

If i have utility functions that emit logs , how do I get them mirrored in the orion cloud UI? These are functions used inside of tasks, not tasks themselves--so no access to

get_run_logger

. I've seen some references to

PREFECT_LOGGING_EXTRA_LOGGERS

, but no legit examples for code that isn't library code. I have code like this at the top of all my utility function modules

logging.getLogger("utils")

and have set

PREFECT_LOGGING_EXTRA_LOGGERS='utils'

to no avail. Logs don't get streamed up to prefect UI

Hi all, I am trying to use string block and fetch the string value string_block_from = String.load("from-string") It returns this String(value='2022-07-01T000000'), how can I only get the value ie '2022-07-01T000000'?

Hi all! I am using Prefect 1.0 and looking for a way to update the flow dynamically. From this doc: https://docs-v1.prefect.io/api/latest/core/flow.html#flow-2, I see the function

set_dependencies

I am trying to have this example:

@task(name="task_A")

def task_A() -(list, int):

test_output_list = list()

test_output_int = 10

return test_output_list, test_output_int

@task(name="task_B")

def task_B(input_a: list, input_b: int, input_c: float):

pirnt(f"{input_a}, {input_b}, {input_c},"

with Flow(name = "test_flow") as flow:

ret_a_list, ret_a_int = task_A()

task_B(ret_a_list, ret_a_int, 10.05)

I would like to change the implementation for the

with Flow() as flow

to

test_flow = Flow(name="test_flow")

test_flow.set_dependencies([task_A, task_B], upstream_task=[task_A], downstream_tasks=[task_B])

I have 2 questions: - is the

set_dependencies

the correct function I should use? - how to use

set_dependencies

correctly to pass the output from

task_A

to

task_B

? Thanks

Hi! We are considering Prefect Cloud for our company's data platform and had a few questions regarding authentication: • The Pricing page says the Enterprise plan supports "SSO". Which mechanism specifically? SAML and/or OIDC? And is it possible to use any provider for these or is there only a defined set of providers available? • Is it possible to provision users by some automated mechanism, e.g. SCIM for users from AD? • What about service accounts? Is there an API for creating these?

Is it possible to add a volumeMount via json patches in prefect 2? I keep running into path error, when i look at the kubernetes job source code the shortcut for env looks like:

"path": "/spec/template/spec/containers/0/env/-"

Which means the path for the volume mount should be:

"path": "/spec/template/spec/containers/0/volumeMounts/-"

When i use that path it doesn't work, any ideas?

I am new to prefect but i was going to try and use it to test run my DBT commands. I have DBT setup in a folder locally with a profiles.yml and a project dir however, I cant seem to get the flow to find it.

from prefect import flow

from prefect_dbt.cli.credentials import DbtCliProfile
from prefect_dbt.cli.commands import trigger_dbt_cli_command

@flow
def trigger_dbt_cli_commands_flow():
    dbt_cli_profile = DbtCliProfile.load("dbtbase")

    trigger_kwargs = dict(
        profiles_dir="../dbt/data-dbt",
        project_dir="../dbt/data-dbt/dbt-project/",
        overwrite_profiles=False,
        dbt_cli_profile=dbt_cli_profile,
    )

    trigger_dbt_cli_command(
        "dbt deps",
        **trigger_kwargs
    )
    return result

trigger_dbt_cli_commands_flow()

👋 hullo! Are there plans to add additional functionality in the

prefect-snowflake

library to wrap snowpark APIs?

I have been running agent with prefect 2.0 cloud, but my agent always raise exception after roughly 2 days. Here is the error message:

raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__

prefect.exceptions.PrefectHTTPStatusError: Client error '403 Forbidden' for url

Does anyone know how I can fix it, any help would be much appreciated!

where can I find some examples of how to pass required flow parameters for prefect v1? when i go to register a flow with required parameters, I get

AttributeError: 'Parameter' object ...

. Why is the flow trying to run when I register it? the more unclear thing is that when i declare a parameter, it returns a parameter object when i really just want the value that's passed to the flow

So in Prefect 2, task objects have a handle

submit()

method that lets you create task objects and them submit them directly, using

wait_for

to specify dependencies. Is there any similar mechanism for sub flows? I am interested in launching a "subflow" but without directly calling the annotated flow method, because I don't know which method I'm calling until runtime. For example, I would like to do this:

result = make_rpc_call()
newflow = Flow(
   name=result.name,
   parameters=result.parameters,
).submit(wait_for=[...])

is this possible?

Hey all, wondering if anyone has any idea on how to generate/write a flow dynamically in 2.0? e.g add tasks + dependencies from another graph object… In 1.0 it used to be possible to just instantiate a flow object and then add tasks / dependencies to the flow object (e.g whilst iterating over a dict or a networkX graph), however now flows are decorators on functions I’m struggling to see how I’d do that? I’ll attach some example code in the 🧵 that shows what I’m trying to achieve. Thanks for any suggestions in advance!

Hello, I have another question. It is possible to ensure that no tasks start running unless all tasks have been successfully registered? For example, if I have some code registering tasks:

1   A = Task(...).submit()
...
5   B = Task(...).submit(wait_for=[A])
...
9   _ = Task(...).submit(wait_for=[B])

If an except is throw after line 1 and before line 9, is there an easy way to ensure that Task "A" does not start?

Hello, I have installed Prefect 1.4.0 on a Minikube instance using Helm. I am providing my own values.yaml file and setting the prefectConfig section to switch the backend to server. However, the graphql deployment config.toml still shows backend cloud once deployed. What am i missing here? Thanks!

I have a workflow that uses Prefect 2 and is run from inside a Docker container. I am able to run the workflow, bring up the Orion UI, and see the workflow in the UI with the states changing as expected. However, I am stuck at trying to create a deployment, with an error saying (full traceback in 🧵):

ConnectionRefusedError: [Errno 111] Connect call failed ('0.0.0.0', 4200)

The Orion UI comes up just fine when I do not try to build a deployment with:

prefect deployment build /app/my_package/flows/my_python_script.py:my_flow_name --name generate_deployment --tag dev --infra docker-container

Any ideas what I could be missing? More details in the thread.

Sorry guys, another question: do you have prefect 2 base image based on arm64 architecture. I had a look of the previous discussion in the forum, but no success. Any guidance would be very helpful

Hi all, I’m trying to add headers (basic auth) to the http request to the orion client. I have a python script to deploy flows and blocks. How am I supposed to update the client with httpx settings in my script ? thanks

Hi- could anyone point me to docs describing how to configure a docker agent to pull from a private registry with 1.x? I've run into issues pulling from our gitlab registry and using pre-pulled images with the gitlab registry prefix: myregistry/image_name:tag

Hi- I have a simple question... how are you/what do you all recommend for organizing your work in prefect? Like, should you/do you have one workspace with all the environments in their in a naming scheme with their own agents/work queues or should you/do you have multiple workspaces for each environment (dev, stage, prod, etc.)? Thank you all for your input!

Has anyone seen Prefect throw an error when attempting to run ASYNC tasks that crashes and says 'got Future <Future pending> attached to a different loop'?

has anyone found any sort of python client to run a flow deployed on prefect cloud (2.0) ?

Hi. Very noob question but can't get my head around it. So if we do a Docker deployment, does that mean that we need the orchestration layer running on some other server as well? Can we make that Docker deployment self-running after initiation? The use case: we have a data analytic product with several Docker containers in it and we initiate all using a Docker compose. (We install this in each individual customer premises). Need to use Prefect as our transformation layer (we currently use cron tabs) and it's good if I can make a single Docker container dedicated to Prefect so that it does periodic transformations by itself. Is this achievable with Prefect? Any tutorial will be much appreciated. Thanks p.s.: we evaluated Airflow for this but decided that it introduce too many docker containers for a simple set of tasks

Hello I am creating prefect work queues pipelines for training and sample prediction. I finished training pipeline queues but I don't know how to trigger action for sample prediction pipeline. Cloud you please give resources or docs for that? Or Should I add sample prediction method in training flow? I use prefect 2.0 . Thank you

Is there anyway to add some formatting to a flow's description? I've tried adding line breaks but appears to remove them in the UI. I'm using a self hosted version of Orion. Example description:

description = """
    This is an example description
    \n\n\n
    Schedule: Every 3 minutes
"""

How it looks in the UI:

Hi. DaskTaskRunner is logging dashboard url after cluster is created. Is it possible to stream this log to the Cloud logs? Thanks.

Hey dear community, I've a funny little use case: 1. My flow does some data deduplication 2. Input data are stored in a cached database and marked as read when they were successfully worked on 3. In the final step, the flow checks for already existing data in an external system and I update some external systems with the deduplicated data 4. The flow uses tasks for each of the tasks at hand and the last step is run with the concurrent task runner, meaning that it spawns up to 50 tasks which execute the upsert operations concurrently 5. The flow should run once per 10 minutes (I'm aware that this is a non-ideal setup, but let's say... a lot of legacy SW being involved... 😅) So now the problem: Everything works fine for as long as the flow runs once at a time. As soon as the flow runs twice at the same time, all hell breaks loose 🔥 🔥 🔥 (or at least I potentially end up in having duplicates). A flow running twice at the same time might happen, when one of the flow runs takes longer than 10 minutes and a new scheduled run get started. Is there sort of an option to limit flow runs to one at a time and keep new runs late until all previous runs are completed/failed/crashed (done in general)? Alternatively, what are some recommendations for how to implement such a behavior on the flow side? (At the start of the flow, I could for example check for an existing, active flow and than simply not continue but end the flow before doing any harm - but this somehow feels a little hacky...) I can't really use task concurrency limits here, as in the final step (where I check for existing data and than upsert), I really need to have the concurrent tasks, otherwise the operation is too slow.

Hey! I moved to prefect 2.5.0 from prefect 2.0b11 and got an error with flow deployment. In my app architecture, I want to deploy flows to prefect server from different service with python, and when I do that, this error occurs. Can you help?

I'm trying to get all the flow runs in Running state with client.read_flow_runs() but got error when initialize the FlowRunFilterState object. I couldn't find any example to do this from discourse, does anyone know what I did wrong?

import asyncio
from prefect.client import get_client
from prefect.orion.schemas.states import StateType
from prefect.orion.schemas.filters import FlowRunFilter, FlowRunFilterState, FlowRunFilterStateType

async def main():
    client = get_client()
    state_type_filter = FlowRunFilterStateType(any_=[StateType.RUNNING])
    state_filter = FlowRunFilterState(state_type_filter)
    flow_filter = FlowRunFilter(state_filter)
    flow_runs = await client.read_flow_runs(flow_filter)
    print(flow_runs)

if __name__ == '__main__':
    asyncio.run(main())

Traceback (most recent call last):
  File "get_client_context.py", line 21, in <module>
    asyncio.run(main())
  File "/usr/lib/python3.8/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "get_client_context.py", line 9, in main
    state_filter = FlowRunFilterState(state_type_filter)
  File "pydantic/main.py", line 333, in pydantic.main.BaseModel.__init__
TypeError: __init__() takes exactly 1 positional argument (2 given)

Hi. I have a Prefect 1.0 flow using ECSRun Task. It as originally registered with a memory value of 2GB and created successfully in ECS. Later, we changed the memory to another value and re-registered, but the Task Definition is not changed. Is there anything obvious that we might be missing?

Does anyone know if it's possible for subflow runs to inherit tags from its parent flow run? Our use case is that we're starting a flow run programmatically from a web application and we're currently tagging that flow run with application-specific metadata to make it easier navigate and find flows that are specific to a particular account, for example. The application is multi-tenant so we will eventually have several flow runs executing at the same time and not being able to easily navigate the flow runs page would be annoying.