Hi everyone, I can see that Flow can have multiple schedules, but the schedule on/off toggle attached to the Flow. Is there any way to pause only 1 of multiple schedule?

Hello Team, Hope you guys are doing good. By using SnowflakeQuery we can pass one query per task, Is there any way to create task for Snowflake sql file.

Hello everyone! I'm currently playing around with the artifacts. I'm wondering if i can give artifacts a custom name. does anybody know if it's possible and how to do it?

Hello! I am trying to configure logging that will be integrated with AWS Cloudwatch. I currently have standard logging but I am trying to figure out how to set up the logging so that only ERROR and CRITICAL logs get passed to AWS Cloudwatch. I still need the standard logging but do not want Cloudwatch to include all logs. I know i can add extra loggers but I am confused how to set up both standard and error/critical logging together

Hi everyone I have 2 scheduler time in a flow, how to show all scheduler time via graphql? Because when i run this graphql, only show the read-only scheduler.

{
  flow{
    id
    name
    schedule
  }
}

And the result is:

{
        "id": "86f5fb6f-26c9-4ffe-9961-f28e5538b53b",
        "name": "spark",
        "schedule": {
          "type": "Schedule",
          "clocks": [
            {
              "cron": "0 1 * * *",
              "type": "CronClock",
              "day_or": true,
              "labels": null,
              "end_date": null,
              "start_date": null,
              "__version__": "0.14.13",
              "parameter_defaults": {}
            }
          ],
          "filters": [],
          "or_filters": [],
          "__version__": "0.14.13",
          "adjustments": [],
          "not_filters": []
        }

Hi experts, I am working on using startFlowRun to allow parent/child relationship flows. I need to pass the result from one child flow to another like the following example :

@task
def get_parameters():
	.....
	return {}
	
flow_a = StartFlowRun(flow_name="flow_a", project_name="sample", wait=True)
flow_b = StartFlowRun(flow_name="flow_b", project_name="sample", wait=True) 



with Flow("parent-flow") as flow:
    date_to_process = Parameter("date_to_process", default=None)
    vp = get_parameters(date_to_process)   

    
    flow_a = flow_a(upstream_tasks=[vp], parameters=vp)
    flow_b = flow_b(upstream_tasks=[flow_a])

I was able to do these successfully; but I am NOT sure how I can use the parameter in the dependent flows. In this case, I have a get_parameters() task to compose the parameters that I need to pass down to dependent flow_a. How do I receive/use the parameter I passed in from

flow_a = flow_a(upstream_tasks=[vp], parameters=vp)

? Let's just say, I want to print out the parameters I pass into flow_a. What do I need to do in flow_a.py to reference the passed in parameter?

Has anyone had the problem when a scheduled flow, just stop scheduling?

weekly = CronSchedule("15 1 * * *", start_date=DEFAULT_START_DATE)
with Flow("Daily Extract", schedule=weekly) as flow:
   ...

This flow was happily running at the scheduled time at 1:15 a.m. every day, except now it just stopped getting scheduled. Curiously today is the first day of the next month....

How are mapped tasks billed in new model? As one task or as the size of the mapped task list?

What scopes does the Github personal access token need to have for using the Github storage for registering flows in Prefect Cloud?

Hello, Prefect, this morning Perfect-cloud seems having a glitch and didn’t create jobs on schedule. We have two flows scheduled at 9:20AM didn’t get its run created until 9:29AM PST. please see thread for details.

Hey there! I noticed an extensive conversation around using the

ECSRun

run configuration, with

ECSAgent

and

Docker

storage. I am running into a similar issue - not entirely sure if there is a workaround currently, but here is the error I cannot get past:

[2021-03-31 15:09:51,907] ERROR - agent | Error while deploying flow: InvalidParameterException('An error occurred (InvalidParameterException) when calling the RunTask operation: Task definition does not support launch_type FARGATE.')

My setup:

# flow.py

import prefect
from prefect import Flow, task

@task
def say_hello():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Got here!!!!")

with Flow("Test") as flow:
    say_hello()

--

# deploy.py

flow.run_configs = ECSRun(
    run_task_kwargs={'cluster': 'my-cluster'},
    execution_role_arn='arn:aws:iam::{ACCOUNT_NUMBER}:role/my-ecs-task-role',
    labels=['ecs']
)

flow.storage = Docker(
    env_vars=config.ENVIRONMENT_VARIABLES,
    extra_dockerfile_commands=[
        f"RUN pip install -e /service",
    ],
    files={f"{os.path.join(os.path.expanduser('~'), 'project')}": "/service"},
    image_name=config.DOCKER_IMAGE_NAME,
    image_tag=config.DOCKER_IMAGE_TAG,
    registry_url=config.DOCKER_REGISTRY_URL,
)

Here is how I am running the agent:

prefect agent ecs start --token {PREFECT_RUNNER_TOKEN} --cluster my-cluster --label ecs --launch-type FARGATE

My issue is, I have no idea if I am doing something wrong on the AWS side of things or not; I am trying to wrap my head around the root cause. I’m also noticing there are

task_definitions

and

task_role_arn

parameters and I’m unsure if I am expected to use them or not, where to set those up. Any insight would be extremely appreciated!

Hi there prefect people! Here is a really weird issue which didn't seem to happen in the past, but now we've noticed it and we're not sure what to think of it. Here is a simple flow with simple tasks and mapping, but somehow the mapped output in the last task has a different length than the input, but everything is successful in the flow. For example, if you check the image below, you can see the difference in number of outputs, but this is unexpected... Here is the MWE for the simple flow: https://pastebin.com/1hGeYHjV

New Prefect user here. I’m trying to understand the difference between static and dynamic Task arguments, as mentioned in this user warning: https://github.com/PrefectHQ/prefect/blob/553544ca25e454140dda363c8f8ca648059a874c/src/prefect/core/task.py#L154-L156 The warning points to a part of the docs that mentions neither the word “static” nor the word “dynamic”, so I’m not sure what the authors really mean when they use these words.

Hello, I d like to use a custom function for a filter in a schedule's clock. using my own lambda or script defined function throws an error of serialization: prefect/utilities/serialization.py", line 397, in _serialize raise ValidationError("Invalid function reference: {}".format(value)) marshmallow.exceptions.ValidationError: Invalid function reference: <function mon_sat_day_filter at 0x7f35e22d78b0> Is there a way around this ? Is there away to interpret a list of filter funcs with OR condition rather than AND ? For reference, Iam trying to run a job Mon - Sat, so an OR condition on a list of is_day_of_week filter func would do. Thank you!

Hello! Is there a way to search for a flow run by the run’s parameters? Either through the UI or through GraphQL

Hello, I'm not sure if my account is acting up or if there's a bug, but I'm unable to delete a secret in my account with the name

test/

. I was trying to see if a secret could be named

test/this/path

and after adding it

test/

was the only part that stayed. Now it can't be deleted.

So, I have a flow that often hangs and doesn't "complete" even when the Reference tasks is done. I see this error message come up in the logs:

RuntimeError: cannot schedule new futures after shutdown

Not sure what to do? I don't want to manually have to set its state to "Success"

I'm having trouble running a simple "hello world" flow on a temporary dask

EC2Cluster

. I'm new to dask. The relevant part of my flow script is:

flow.executor = DaskExecutor(
    cluster_class='dask_cloudprovider.aws.EC2Cluster',
    cluster_kwargs={'n_workers': 2, 'docker_image': 'prefecthq/prefect', 'debug': True}
)
flow.run()

This fails with

FlowRunner: ClientError('An error occurred (InvalidParameterValue) when calling the RunInstances operation: User data is limited to 16384 bytes')

. The issue is that the docker run command looks like this:

docker run --net=host   prefecthq/prefect env DASK_INTERNAL_INHERIT_CONFIG="a_very_very_long_string" python -m distributed.cli.dask_scheduler

, so I guess the command winds up being too long. Does anyone know a workaround for this issue?

Hi everyone, I have a question I have 2 mysql db (A & B) and want to ingest it to bigquery. I only want to have 1 Flow -> mysql_to_bigquery I planned to use multiple schedule, schedule A with db A default parameters, schedule B with db B default parameters. And next, I want to these 2 db ingested with different KubeAgent by label. Is there any way to assign label to the schedule?

Hi Prefecters, does anyone know the command to delete a flow? Or can it be done only using the UI?

Hi everybody, i am wondering for the

transform task

, it is expected to fail for parameter

"a"

, but when i rerun from fail, what is the expected flow? Will it rerun for all cases

[3,1,"a"]

or skip

3,1

, and only rerun

a

@task
def extract():
    return [3, 1, "a"]


@task
def transform(x):
    y = x + 1
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(y)
    return y


with Flow("failure-flow-test", executor=LocalExecutor(), run_config=LocalRun()) as flow:
    e = extract()
    t = transform.map(e)

I’ve got a really strange thing happening with Cloud: some of our schedules have stopped working. 1. Schedules that have been setup in code were working fine for months 2. Last week we setup one more schedule in Cloud directly 3. Now only the schedules that we setup in Cloud are working

If you are using results e.g. S3Result, how do you debug the content of the generated files without actually running the flow that is using them?

Hello! I have a question. How to get result of

Resume

state passed via

Client.set_task_run_state

method to paused task in task runtime? When i try to use

Client.get_task_run_state

in task i found that state in task runtime is

Running

(after

Resume

and after

Submitted

). So passed result to

Resume

state is not accessible. Have any ideas?

Short and sweet I hope. But it isn’t clear to me from the docs where data is cached output data is persisted when using Cloud backend. I know locally it’s stored in memory, and you cannot use the cache locally unless you make use of a backend API, but does this mean that the output data is actually stored in the backend - Cloud or Core server? If so, does this not violate the principle that no data flow data should be stored on the API side?

where can I find a document on detailed feature comparison for each different Prefect package(starter, standard and Enterprise)? https://www.prefect.io/pricing

Hello, what is the best way to use global variables, for example database engine objects, in different tasks?

Hi, all, can you help me, i need a sequential execution on map method (not parallel, but serial), or something similar to achieve sequential looping over a list of params. Here's my code:

from prefect import task, Flow, Parameter
from prefect.executors import LocalDaskExecutor, DaskExecutor
from prefect.tasks.prefect import StartFlowRun
import prefect

executor = LocalDaskExecutor(num_workers=1)


flow1 = StartFlowRun("flow1", project_name='test', wait=True)
flow2 = StartFlowRun("flow2", project_name='test', wait=True)

with Flow("example", executor=executor) as flow:
		
	table_dict_param_list = []
	with open('parameters.txt', 'r') as f:
		lines = f.readlines()
		for cnt, line in enumerate(lines):
				dict_param = {}
				dict_param['param1'] = cnt
				dict_param['param2'] = line
				table_dict_param_list.append(table_dict_param)
	
	flow1 = flow1()
	
	## i need this execution to be sequential like in a "for loop"	
	flow2.map(parameters=table_dict_param_list)

flow.register(project_name="test")

Design question: Why isn’t

idempotency_key=flow.serialized_hash()

the default? https://docs.prefect.io/orchestration/concepts/flows.html#core-client I can’t think of why someone would want the Flow version to change if the Flow definition hasn’t.

Hello all, I'm migrating to the newly released versioned Prefect Server helm chart and deploying it on GCP, but I seem to be running into a couple deployment issues.

prefect-server-towel

seems to have an

ErrImagePull

error, and

prefect-server-hasura

and

prefect-server-ui

can't pull their respective images from the registry. Here's my deployment command I entered:

helm repo add prefecthq <https://prefecthq.github.io/server/>
helm install ${NAME} prefecthq/prefect-server --values=values.yaml