hi, I've found the conditional tasks to be rather confusing. In particular I was hoping that if you have something like this...

a = task_a()
test_value = task_get_test_value()
with case(test_value, 'test-against-this'):
  b = task_b()
  b.set_upstream(a)

...then task

b

would fail if

a

fails regardless of whether the

case

block is in the skipped or non skipped state. However that doesn't seem to be the way it works - rather if the

case

block skips, then it 'hides' whether or not

a

succeeded.

I am trying to run a task after an apply_map, but I am unsure how to link it up. It keeps running first with no upstream tasks.

with Flow(
    "AWS CUR import",
    result=GCSResult(bucket="platformsh-vendor-prefect"),
    state_handlers=[flow_failure],
) as flow:
    # save_file = prefect.Parameter("Download Files", default=False)
    num_periods = prefect.Parameter("num_periods", default=4)
    periods = generate_periods(num_periods)
    apply_map(run_or_bail, periods)
    clear_files()

I am experiencing an interesting issue with

upstream_tasks

, I have the following example flow:

with Flow(
    'example',
    executor=LocalExecutor(),
) as flow:
    data_csv_url = Parameter('data_csv_url')
    output_url = Parameter('output_url')

    data = generate_data(data_csv_url)
    data_with_stuff = generate_stuff.map(data)
    data_with_stuff_written = write_data(data_with_stuff, output_url)

    data_with_other_stuff = generate_other_stuff.map(
        data_with_stuff,
        upstream_tasks=(data_with_stuff_written,),
    )

I read a csv file and generate some “data” objects (a list of them) for the given flow run. I think map them across another task, which mutates the data items to add things to them. I want to then write the state of the objects to a file for tracking and I want to do this BEFORE I make further modifications in`generate_other_stuff`. With this flow def,

generate_other_stuff

fails with out any exception and does not seem to even run the task

ex

└── 09:09:26 | INFO    | Task 'generate_data[0]': Starting task run...
└── 09:09:26 | INFO    | Task 'generate_data[0]': Finished task run for task with final state: 'Success'
└── 09:09:26 | INFO    | Task 'write_data': Starting task run...
└── 09:09:26 | INFO    | Task 'write_data': Finished task run for task with final state: 'Success'
└── 09:09:27 | INFO    | Task 'generate_other_stuff': Starting task run...
└── 09:09:27 | INFO    | Task 'generate_other_stuff': Finished task run for task with final state: 'Failed'

Hi folks, is there a way to make a manual_only task get trigger failed and not pause the flow run in case there is an upstream task run failure ?

Are Prefect artifacts going to become a fully fledged feature in the near term? I see they are currently in an experimental phase.

Hi Prefect Team, What version of Nginx does the last Prefect core 0.15.6 use?? I'm asking cause the prefect core 0.14.8 uses Ngix 1.18.0 (with docker) which has a vulnerability that is patched in Nginx 1.20.1.

Hi community, i am trying to deploy prefect on google kubernetes and i am unable to do it. i am using the command mentioned in documents prefect agent kubernetes install -k API_KEY | kubectl apply --namespace=default -f - and when i see the pods using kubectl get pods there in status of prefect agent it says CrashLoopBackOff. When i log the pod i see the following error prefect.exceptions.AuthorizationError: [{'path': ['auth_info'], 'message': 'AuthenticationError: Forbidden', 'extensions': {'code': 'UNAUTHENTICATED'}}]

Hi, I have been trying to get a prefect agent running on kubernetes, starting with locally on minikube.

This message was deleted.

Hey there. We have been seeing this error during some flow runs

No heartbeat detected from the remote task; marking the run as failed.

. Any info on what is possibly causing this?

Hello everyone! In a production environment that we have, we do not have access to the prefect UI and need to use the

prefect

CLI tool to do things like check which flows are running and see their logs. If a flow is running and I want to cancel it, how could I do this? There doesn't seem to be a command, and I'm assuming I need to do it with graphql. Is there an example of how to do this anyway? I appreciate it in advance.

hi all, Does prefect have integration with talend? thanks.

#prefect-community Hello Everyone, I would like to know if Prefect is compatible with Windows Containers

Hello there! I am currently getting much of our codebase to work with prefect. In that regard I am doing some flows, where I use the Mapping feature for processing data. I am currently running into a problem, where I want to output multiple values from a mapped task, but I cannot seem to get it working. I have tried specifying both output annotation and nout=2, but it does not seem to register. It returns a TypeError exception saying that the task is not iterable. I have another task which is not a mapped task where this works fine. Is there anyone that have tried this before and can help me get it working? Thanks in advance!

Hi there I'm still struggling with the concept of pickle based storage -> we use s3 as storage and docker run. I read the docs here: https://docs.prefect.io/orchestration/flow_config/storage.html#pickle-based-storage). What exactly is persisted in pickled file? • For sure it's the flow-object • but also global variables are frozen (e.g. datetime.now() will be never refreshed in the run, even it's called outside of the flow-context) • but every piece of code in a task will be executed So if I build some objects while the flow build / register, this objects are frozen. Right? But if I change some methods for these objects, rebuild the image and rerun the flow based on the "previous" flow-definition the new method will be called (but not if I change the init-method). Is there any explanation which parts are dynamic and which part are frozen. Thanks for any help!

Hi! Im trying to use The Great Expectations task in our Prefect workflow running on prefect server in kubernetes. I'm struggling with providing a data context. Since the ge.data_context.DataContext object can not be serialized with cloudpickle, I cannot initialize it in one task and pass it on to the RunGreatExpectationsValidation task. Any one knowing of a solution to pass such an object between tasks?

Hi! I am trying to run prefect orion, but I am getting this error Can you please help? I have tried it on WSL, Windows10, linux. I am getting the same error on all three platforms. $prefect orion start $prefect version 2.0a3

Hi, i have been trying to deploy prefect on Google kubernetes, I have generated manifest file using command prefect agent kubernetes install --rbac -k my_key so when i try to deploy using skafold file build fails every time with this error Starting deploy... - deployment.apps/prefect-agent created - Error from server (Forbidden): error when creating "STDIN": roles.rbac.authorization.k8s.io is forbidden: User "468642734279@cloudbuild.gserviceaccount.com" cannot create resource "roles" in API group "rbac.authorization.k8s.io" in the namespace "default": requires one of ["container.roles.create"] permission(s). - Error from server (Forbidden): error when creating "STDIN": rolebindings.rbac.authorization.k8s.io is forbidden: User "468642734279@cloudbuild.gserviceaccount.com" cannot create resource "rolebindings" in API group "rbac.authorization.k8s.io" in the namespace "default": requires one of ["container.roleBindings.create"] permission(s). kubectl apply: exit status 1

Hello everybody! I am actually trying to test a PoC flow using Docker storage and DaskExecutor. I deployed a local prefect server and a separate dask cluster . Runing the flow directly works well but when registering and running using a

prefect agent docker

i am having a permission error : Thanks for your help

Hi all. Trying to do a backfill on one of my flows. The flow takes in a DateTime parameter and all subsequent tasks use that to crawl/transform data.. so I do something like:

for target_date in target_dates:
    f.run(parameters={"target_date": target_date}, run_on_schedule=False)

I am running the above using a

LocalDaskExecutor(scheduler="processes", num_workers=12)

locally. The problem is that run time grows...

Took 15.72 sec for 2016-05-08
Took 20.59 sec for 2016-05-09
Took 32.52 sec for 2016-05-10
Took 43.89 sec for 2016-05-11
Took 55.81 sec for 2016-05-12
Took 67.15 sec for 2016-05-13
Took 77.92 sec for 2016-05-14
Took 90.59 sec for 2016-05-15
Took 105.21 sec for 2016-05-16
Took 115.50 sec for 2016-05-17

Would that be a dask issue? Any suggestions for maybe running the backfill locally in a different way welcome! (There is no back aggregation involved, so no reason for later runs to take longer. Volume/size of data is almost identical for each run. So when a backfill completes and I start another one fora different set of dates the time profile is similar: very fast -> slower)

If a task in a flow failed but the flow hasn't failed yet (because there are other branched tasks in the flow run that are still running), how can I restart the failed task without having to wait for the other tasks to finish?

hello! Am I able to attach cloud hook to all flows in project? or group flows to one cloud hook somehow

Hello! Is it possible to set a template name (using target) for a task with multiple return values? I'm doing something like this:

@task(checkpoint=True, target="{task_name}.pkl", nout=4)
def split_train_test(dataset):
    array = dataset.values
    X = array[:,0:4]
    y = array[:,4]
    X_train, X_validation, Y_train, Y_validation = train_test_split(X, y, test_size=0.20, random_state=1)
    return X_train, X_validation, Y_train, Y_validation

Using LocalResult on my flow, I noticed that I've saved an

split_train_test.pkl

, which was what I expected, but also 4 files with prefect default names (

prefect-result-...

). Is it possible to define a template for these files too? Or just not save them separately? This is important because, with these names, the cache will never work for them (as it have the current timestamp)

As part of our CICD we have a script that registers flows for us; I've noticed that our script ends up generating new versions of flows even though nothing changed about the flow. So, in order to debug this, I wrote some debugging statements during CICD:

if not args.dry_run:
        project = args.project
        logger.log(SUCCESS, f"Registering flow '{flow.name}' to project '{project}'.")
        logger.debug(f"Serialized flow: {json.dumps(flow.serialize(), indent=4, sort_keys=True)}")
        logger.debug(f"Hashed flow: {flow.serialized_hash()}")
        flow.register(
            project_name=project,
            idempotency_key=flow.serialized_hash(),
            set_schedule_active=set_schedule_active,
        )

I noticed that on back to back execution of the above script, for the same flow, to the same project, different hashes were being created. When I look at the

serialized_hash

of those flows I see no difference. To confirm the serialized flows are the same, I've saved the logging output to two json files (see screenshot). The only difference I can spot is that is use

indent=4

in my json.dumps however the codebase does not.

Hey! I was wondering if anyone had modified the helm chart to use Ingress instead of LoadBalancers for the GraphQL and Apollo endpoints and whether you ran into any issues with it.

Hi everyone! I'm a noob to Prefect, working with cloud, and I'm wondering how to grab one of my secrets saved in my prefect cloud team's secrets while I'm testing locally

I haven't found a code example that actually reaches into a cloud team and collects a secret. It looks like "use_local_secrets" needs to be passed to an argument, but I'm not sure where. For now, I just want to set this stuff for my current python session.

Can anyone help me out?

Hi, does someone knows or recommend any material/course/book that uses prefect?