Also what is the best practice for sharing code between flows in a project? And how about sharing code between different projects? Are there any options besides rolling out a library to a pypi index?

Is it possible to set the default value of a Parameter to be a lambda instead of a value? I want to re-create something like the flow names generated by the cloud ui.

I have a question about my (probably incorrect 😄 ) understanding of Prefect logger and it’s behavior when using a

DaskExecutor

. When running this example Flow

@task
def say_hello():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Hello, Cloud")
    return "hello result"


with Flow(
    "dask-test-flow",
    storage=storage.S3(
        bucket=storage_bucket
    ),
    run_config=ECSRun(
        image=worker_image,
        labels=json.loads(os.environ["PREFECT_AGENT_LABELS"]),
        task_definition=definition,
    ),
) as flow:
    hello_result = say_hello()

I am able to view log information (specifically the “Hello, Cloud” info line) directly in Prefect Cloud. But using an identical flow with a

DaskExecutor

logging information is sent to my CloudWatch

dask-ecs

log group but is not available in Prefect Cloud. Is it possible to view all Dask worker logging in Prefect Cloud and if so, what additional logging config to I need to include?

Good day! When we use dask-cluster, is the total number of concurrently executed tasks equal to the number of dask-workers or the sum of dask-worker threads?

hi everybody, i saw this in the prefect document

from prefect import Task, Flow
from prefect.tasks.prefect import StartFlowRun


class ExtractSomeData(Task)
    def run(self):
        return {"param-key": "some-random-piece-of-data"}

extract_some_data = ExtractSomeData(name="extract_some_data")

# assumes you have registered a flow named "exampled" in a project named "examples"
flow_run = StartFlowRun(flow_name="example", project_name="examples")

with Flow("parent-flow") as flow:
    flow_run.set_upstream(extract_some_data, key="parameters")

but i am wondering what is the

parameters

? and whether the

param-key

from

ExtractSomeData

is passed into the

example flow

hello one and all I’m trying to analyse the memory consumption of one of my flows, but I cannot figure out how to add guppy to my flow. My attempt was to create two tasks: one to create the hpy object and another to print it. I added dependencies to make the object creator the first task in the flow and the printer the last task in the flow. The heap is actually printed, but only for the task that created the hpy object. Does anybody know how I can have the hpy() object cover the entire flow? (or if there is an alternative to using guppy, that’d be fine as well)

I'm looking through the

ShellTask

source because I was trying to see if there's a way to stream the task output when the log level is

INFO

. Is it possible? I have a task whose output i'd like to regularly stream to my logs, but I don't want to have the entire flow running in debug mode

Hello all. I am trying to write a simple task and get the return value from that task. I have been trying to follow https://docs.prefect.io/core/concepts/tasks.html#indexing as what is stated seems pretty straightforward. I have also been looking at the "Multiple Return Values" on that same page. Regardless of what I do, I can't get the expected values of 1 for the Indexing example, and the other values from the "Multiple Return Values" example. Rather, when I am printing the results, I get a <Task: xxxxxxx> type of string displayed. As stated above, I feel like this should be straightforward, but I must be missing something.

Hey all! Does anyone know how you can pass kwargs to a task function? Right now I have a task that does a very similar thing with some tweaks depending on keywords a few times.

@task(timeout=60, nout=2)
def get_date(
    dt,
    ensure_explicit=False,
    max_late=timedelta(hours=12),
):
    # some code
    return None, None

However, when I try and add to my flow a

get_date(dt, max_late=timedelta(minutes=1))

I get

TypeError: got an unexpected keyword argument

. I assume the task decorator is wrapping the original function and doesnt pass through any kwargs it doesnt recognise. Is there a subtle keywork Im missing to pass kwargs through, or should I rewrite these functions of mine to utilise class based Tasks?

Hi all 🙂 I was wondering - is it possible to trigger Prefect workflow externally (using Prefect Core deployed on my machine - not Prefect Cloud)? Thanks for your help!

Hi all, Prefect newbie here 🙂. I’m looking for some advice on best practice around sequencing tasks. The tasks aren’t passing any data between each other, but must run in my specified sequence.. I’ve played around with set_upstream/downstream but can’t come up with the schematic I’m after. Use case: Task 1 executes Task 2 and 3 both execute on success of Task 1 Task 4 executes on success of both Task 2 and 3 Any ideas or pointers would be appreciated 🙂

And another one from me, on a completely different topic. It looks like Prefect has good support for AWS Lambda, which is awesome. However, I couldn’t see a way to specify the AWS endpoint to use, only the credentials etc. Am I wrong? Our use case for Prefect is to string a series of Lambdas together, and we’re using serverless framework (serverless offline) in our local development environment to run Lambdas on a localhost endpoint. It would be great if I could get Prefect local talking to my local Lambdas.

Hey guys! I'm new to Prefect as I'm trying new tools. How does Prefect compare to Dagster? Also, would MLFlow complement the use of Prefect?

is there a way to explicitly remove a task from another task's set of upstream dependencies?

What is the correct format for AWS secret environment variables? I tried the following, but it doesn't seem to work:

PREFECT__CONTEXT__SECRETS__AWS_CREDENTIALS={"ACCESS_KEY": "myaccesskey", "SECRET_ACCESS_KEY": "mysecretaccesskey"}

How do you go about setting up a volume for jobs when using the kubernetes agent? I see that

--job-template

is an option when using start, but not when using install. I'm able to set up a volume on the agent, but I realized that won't actually make the volumes available to my flows when I run them.

Is there a way to use multiline parameters in the Prefect Cloud UI?

Does anyone have a docker-compose file that spins off all the services for UI, server, etc ?

Hi all! I have registered a flow without a schedule and can trigger it fine from the Prefect Cloud UI. When I try to create a schedule for it (in Prefect cloud UI) and enable it for the flow, it does not recognise the schedule I have just created. Also if I refresh the page I can see the schedule “disappeared”, like it has never been saved. Has anyone experienced this or is there something I am missing?

Hi Team, I'm evaluating Prefect as data flow pipeline and started learning and trying out. My setup is Prefect Server, LocalAgent running inside a docker and creating task from jupyter Notebook and registering flow using S3 storage. I could all but when I run getting error "Failed to load and execute Flow's environment: AttributeError("'NoneType' object has no attribute 'setup'")"I know making some simple mistake, any help would be appreciated. Flow is simple hello world. Since my prefect server running remote, I couldn't find option to pass API url to flow, hence created client and used to register, registration works. here is my code for that. import prefect from prefect import task, Flow, client from prefect.storage import Local from prefect.storage import S3 clientP = client.Client(api_server="API URL") storage = S3(bucket="bucket",client_options={"aws_access_key_id" : "awsaccesskeyidXXX", "aws_secret_access_key" : "awssecretaccesskey"}) @task def say_hello():     logger = prefect.context.get("logger")     logger.info("Hello, Cloud!") with Flow("hello-flow",storage=storage) as flow:     say_hello() # Register the flow under the "tutorial" project clientP.register(flow,"learning")

how do you verify that a DaskExecutor was used to run a flow? There is no info about the executor in the logs, right?

Hi everyone, when I register a flow with prefect-cloud/prefect-server I can see the task graph on prefect-server on the schematic tab (image shown below) of the UI. I would have thought that dask would calculate that graph. Is that true? If I'm using the dask executor is that graph then recalculated for each flow-run by the dask-scheduler? Or is the task graph calculated by dask at flow registration time and then cached somehow, so it doesn't need to be recalculated? If there is any documentation on this, please direct me there. Thank you!

Hi folks, when trying to visualize a flow that ran locally using

flow.run()

and

flow.visualize(flow_state=flow_state)

is there a way to display the duration of the tasks ?

Is there an idiom for using case logic to define paramters to tasks without duplicating the entire flow inside the case? Something like this?

with case(is_rapids(input), True):
    export_result = ExportsTask()(secret_task, 'somekey', stack)

with case(is_rapids(input), False):
    export_result = ExportsTask()(secret_task, 'anotherkey', stack)

MyTask()(input, export_result['training_job_queue'], export_result['training_job_definition'], PipelineArtifactsBucketTask()(secret_task, stack))

Hi! Quick question, I've upgraded my k8s agent to v0.14.9 and understand based on work here and here that the separate resource-manager has been removed from the k8s agent. I expected the newly deployed agent to clean up successful k8s jobs but finding them hanging around. Anything I'm missing here? Thanks!

I’m trying to find information on how I can expose the address of the Dask scheduler to use the distributed dashboard https://docs.dask.org/en/latest/diagnostics-distributed.html when using the Temporary Cluster approach described here https://docs.prefect.io/orchestration/flow_config/executors.html#using-a-temporary-cluster. The

DaskExecutor

is dynamically creating these temporary clusters https://github.com/PrefectHQ/prefect/blob/05cac2372c57a93ea72b05e7c844b1e115c01047/src/prefect/executors/dask.py#L213 on a per flow basis so I’m unsure how I can obtain the scheduler’s address (and dashboard link) without some hook here in the Prefect code? In an ideal world I would like to have the Dask scheduler’s public address and dashboard link reported as part of the Flow logs in Prefect UI. Any suggestions are greatly appreciated.

Hello all. I’m new to prefect and I’m trying to figure out an optimal way to deploy Prefect on Google Cloud Platform in a serverless architecture. I’ve come across a lot of material for AWS but nothing on GCP. Would be grateful if someone could share. I aim to use Prefect Cloud

I have searched and searched for this one. I have dev/test/prod I write my flow, build into into a docker image and run it on the dev prefect server instance. No worries. What I want to do is promote that docker image and register it with the test and later the prod prefect server.

Prod does not really have much in it. Ideally I'd just like to use the command line to tell prefect server to load up the flow in the existing docker container without the flow present outside of it.

Is there a library that allows you to write flows like this:

chain(
        extract(from_date, to_date, base_currency), 
        transform, 
        load
    )

Instead of this:

extract_result = extract(from_date, to_date, base_currency)
    transform_result = transform(extract_result)
    load(transform_result)

?