Hi, my task is mapped to some children tasks. if some of the children tasks fail in a run, is there a way to skip the successful ones and only re-run the failed? Can this be achieved by state handlers? thanks

If I set up a "flow of flows" orchestration and schedule it, and the flow of flows all runs before the scheduled runs for the individual flows, will the flows just run again (unnecessarily)?

I have a

flow_A

with

task1, task2... task10

. Now I want to construct

flow_B

with

task1, task2.... task 20

and

task11_result = task11(task10_result)

. How do I add the tasks of

flow_A

as upstream tasks for

flow_B

. I know I can do

create_flow_run

, but I do not want monitoring for

task1...task10

in

flow_B

Kind of like:

with Flow("flow_A") as flow:
    flow_A_param1 = Parameter("flow_A_param1")
    ...
    flow_A_param_n = Parameter(...)
    ...
    task1 = ....
    ...
    task10_result = task10(...)

with Flow("flow_B") as flow:
    flow_A_param1 = Parameter(...)
    ...
    flow_A_param_n = Parameter(...)
    
    flow_A_result = flow_A.run(parameters = {"flow_A_param1": flow_A_param1,...., "flow_A_param_n": flow_A_param_n}) #dont want monitoring for these tasks
    task10_result = flow_A_result.result[flow_A.get_task(task10)]
    task11_result = task11(task10_result)
    ...
    task20_result = task20(...)

where `flow_B`'s parameters are just `flow_A`'s parameters. But in the example above I always have to repeat the parameters of

flow_A

in

flow_B

and keep them both up to date with each other. Is there a better way to do this?

👋 Hi everyone! We're working with Gradient Flow to collect some data on workflow orchestration use cases and features. We want to know what's important to our community as we move forward. Please consider taking this 10 question survey. All responses are anonymous, of course.

Is there any chance that Prefect could release a arm64 image?

I am running into some unexpected behavior with extra loggers in prefect cloud. When I add my extra loggers to the config and run my flow with a local agent everything works as expected. When then switching to the DockerRun config and Bitbucket storage my extra loggers no longer get passed to the Cloud Logs. Is there something extra I need to be doing when moving from a local agent to a docker agent to get my logs to be passed through?

I'm using pylint and it seems like when I use the

upstream_tasks

kwarg (e.g. as in example below) I get an

Unexpected keyword argument 'upstream_tasks' in function call (unexpected-keyword-arg)

... is there a way I can get pylint to recognize this is ok? (other than disabling?)

@task
def add_one(val):
    return val + 1

with Flow("Add One") as flow:
    p = Parameter("param")
    first = add_one(p)
    second = add_one(p, upstream_tasks=[first])

How can I fit in these Gitlab's sheetload script in the Prefect workflow? Is there similar example for this use case?

Hi! I want to make sure that only one run of a flow runs at a time, is there a way to restrict the paralell runs of a flow?

Hi, i have a task function which retries multiple times to check if a file is present and if successful starts the downstream task. This same task has different parameters going into it, the ones in orange color are working correctly (going into retries) but few of them are in yellow color (going in submitted state) and causing downstream task to fail. What is this submitted task? And why is it happening ?

Hi, Maybe someone can help on how to do best options in such case 1) We have a linux docker image as Prefect Agent with mounted S3 storage acting as primary storage for Flows, configs and some resources used for tasks. 2) Each Flow is executed in separate container using DockerRun(image=prefecthq/prefect:0.15.13-python3.8, host_config = host_config). So we have isolated Flow Containers inside Agent Container :). host_config is used to mount Agents S3 into Flow runtime, so that Flow, when running can pick up configs and resources mounted to Agent. Tricky thing is how to pass connection strings inside Flow Container? 1) I thought that we can pass them as arguments of DockerRun(.... environment = environment) according to (https://docker-py.readthedocs.io/en/stable/api.html#module-docker.api.container), but we need to prepare the list and it means that it will be "plain text". Not good. 2) Another option is to define some dictionary in the Flow and replace variables in CI/CD tool during deployment, but in this case it will be stored as plain text on storage inside Flow. This might be an option in case if serialized Prefect flow can be encrypted. 3) ??? maybe some other ideas?

Hi there !

I have noticed that my flow got Permission denied for write in home directory. Does anyone knows about this error? File "/usr/lib/python3.9/os.py", line 225, in makedirs mkdir(name, mode) PermissionError: [Errno 13] Permission denied: '/MboiTui' I have changed the permission for my home '/MboiTui' as 777, with chmod.

Hi there, Trying to save a csv to GCS using pandas in a Prefect cloud flow... I have set the GCP_CREDENTIALS key on prefect cloud, which I am beginning to realize is for prefect GCP tasks only, but I am getting the GCS 401 error that suggests the credentials are missing. Is there a way to pass the credentials into the pandas operation using the saved credentials on Prefect Cloud or is there another way using Prefect tasks that could work?

I tried to get the logs of my run to analyse failed runs:

prefect get logs -n name_of_the_run

but they seemed to be incomplete. I get only 2636 lines, for less than an hour of the flow. The logs are there since I see them in the UI. Any suggestions?

Hello. We are working with

prefect[aws]==0.15.3

. We have a flow that generates a large number of batch jobs, then submit them, splits them to chunks of 100 and generates an AWS waiter object per 100. We have one very large jobs with over 450 batch calls. This job seem to be stuck even though it progressed until the end. When tracking the progress we can see that a few tasks are stuck as mapped even though all child tasks are done. i.e. 472 successful definitions created in

create_ive_analysis

then all of them were submitted and chunked in

submit_jobs_analysis

and then all 5

AWSClientWait

jobs were done (we call it with map on the chunked job ids). The parent block is still in mapped mode. Also, sometimes an

AWSClientWait

tasks fails and the job doesn’t fail, it just stays there (again this is with a mapped

AWSClientWait

) Wait code

wait_res = AWSClientWait(client='batch', waiter_name='JobComplete').map(waiter_kwargs=batched_waits)

Where batched_waits is created by

@task
def submit_jobs_and_batch_job_awaits(jobs: List, batched_num=BATCH_CLIENT_WAIT_MAX_SIZE):
    submitted_jobs = [BatchSubmit().run(**job) for job in jobs]
    waits = []
    for i in range(0, len(submitted_jobs), batched_num):
        waits.append(
            {
                'jobs': submitted_jobs[i : i + batched_num],
                'WaiterConfig': {
                    'Delay': 10,
                    'MaxAttempts': 10000,
                },
            }
        )
    return

What could cause that?

Hi guys, How can I get the working directory or file path for Docker Run with GitHub Storage? I need to import some utilities from my repo, but I keep getting import errors.

Hello, I am running an Azure DevOps pipeline with service connection set up to access my AWS account. As a pipeline task, I run a sagemaker processing job that includes a Prefect flow.

- task: AWSShellScript@1
              inputs:
                # awsCredentials: 'xxxxx'
                regionName: 'xxx'
                scriptType: 'inline'
                inlineScript: python3 cli.py -remote S3 -p "['a','2020-09-01', '2020-09-02']"

I get this error:

[2022-02-03 21:29:07+0000] ERROR - prefect.S3Result | Unexpected error while reading from S3: TypeError('expected string or bytes-like object')
Traceback (most recent call last):
  File "/miniconda3/lib/python3.7/site-packages/prefect/engine/results/s3_result.py", line 142, in exists
    self.client.get_object(Bucket=self.bucket, Key=location.format(**kwargs))
  File "/miniconda3/lib/python3.7/site-packages/botocore/client.py", line 391, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/miniconda3/lib/python3.7/site-packages/botocore/client.py", line 692, in _make_api_call
    api_params, operation_model, context=request_context)
  File "/miniconda3/lib/python3.7/site-packages/botocore/client.py", line 738, in _convert_to_request_dict
    api_params, operation_model, context)
  File "/miniconda3/lib/python3.7/site-packages/botocore/client.py", line 770, in _emit_api_params
    params=api_params, model=operation_model, context=context)
  File "/miniconda3/lib/python3.7/site-packages/botocore/hooks.py", line 357, in emit
    return self._emitter.emit(aliased_event_name, **kwargs)
  File "/miniconda3/lib/python3.7/site-packages/botocore/hooks.py", line 228, in emit
    return self._emit(event_name, kwargs)
  File "/miniconda3/lib/python3.7/site-packages/botocore/hooks.py", line 211, in _emit
    response = handler(**kwargs)
  File "/miniconda3/lib/python3.7/site-packages/botocore/handlers.py", line 238, in validate_bucket_name
    if not VALID_BUCKET.search(bucket) and not VALID_S3_ARN.search(bucket):
TypeError: expected string or bytes-like object

the same task runs good when I just run it on an Azure Devops agent:

- task: AWSShellScript@1
              inputs:
                # awsCredentials: 'xxxxx'
                regionName: 'xxx'
                scriptType: 'inline'
                inlineScript: python3 cli.py -local S3 -p "['a','2020-09-01', '2020-09-02']"

has anyone had this issue before?

Hello, hope this would be the right channel for this question... • I already have my tasks working in my local machine, actually right now running my tasks manually, and I wanted to start deploying this tasks to a production env and prefect cloud • This is an ETL common process but with storing files locally to not waste too much RAM among tasks and because the requests I'm doing to extract data are really heavy • So, in the extract process, after I just finish a smaller task, I store the result in a local file • all of this is working in my local machine with no issue with flow.run() • But when I register the flow for prefect cloud the issues start ◦ right now I'm dealing with

FileNotFoundError: [Errno 2] No such file or directory: 'data/file.csv'

◦ this in prefect cloud ◦ mine is basic config. local agent in a server ◦ my code in a directory where I want to create the file with a directory ◦ the line of code is a

pathlib.mkdir

sentence I'm sure there is something I dont understand about prefect, because I cannot manipulate the local filesystem in the server with prefect cloud, any ideas?

Hi there, I’m trying to setup a demo flow package inspired from here, the idea is to keep Gitlab as a source of truth and as a

STORAGE

for the flows and have the Gitlab CI/CD register the flows to the Prefect Cloud. However, I receive an error when I try to run the dummy flow taken from the official docs :

from prefect import task, Flow
from prefect.storage import Gitlab


@task(log_stdout=True)
def hello_task():
    text = f"hello World"
    print(text)
    return text


with Flow(
    "hello-flow",
    storage=Gitlab(
        repo="predicthq/data-engineering/prefect-server",
        path="flows/hello-world/flow_hello_flow.py",
        access_token_secret="",
    ),
) as flow:
    hello_task()

The error:

(.venv) ➜  prefect-server git:(main) ✗ python flows/hello-world/flow_hello_flow.py
Traceback (most recent call last):
  File "/Users/farid/WorkSpace/Github_Repos/dataobs-poc/prefect-server/flows/hello-world/flow_hello_flow.py", line 2, in <module>
    from prefect.storage import Gitlab
ImportError: cannot import name 'Gitlab' from 'prefect.storage' (/Users/farid/WorkSpace/Github_Repos/dataobs-poc/prefect-server/.venv/lib/python3.9/site-packages/prefect/storage/__init__.py)

Looking at the source code, I don’t see anything that could cause a dependency circle. Any idea what it could be causing this? Needles to say that I have installed the gitlab package:

pip install 'prefect[gitlab]'

With Orion will we have the option to say we want this flow to run all the time ? The idea would be for that one to listen to event and start other child flow. However this flow will be in a continuous loop.The main question will be how do you release a new version cleanly as the current event loop will need to be stopped and a new instance starter

Hi, I'm looking for a workflow management solution that we can use now to replace our homebrew scripts. I've seen Orion which looks like it would tick our boxes, but I'm aware it's not ready for production use yet. So I have two questions about prefect core: 1. Is it possible to have cyclic workflows? I have a flow which transforms input records into a smaller number of output records. I'd like to then process the output records with the same flow again until it outputs no records. 2. Is it possible to have event-driven flows? We've got records coming into a queue which ideally I'd process when the queue size reaches a certain size. I have read the PINs and suspect the answer is no, but just wanted to check. I can work around it by periodically scheduling a flow that checks the size of the queue before exiting if it's too small

Hello everyone. We have a prefect-server instance running in our cluster, and we were wondering what would be the best way to organize the projects? Is it possible to create directory-like structure of the form

{DOMAIN}/{PROJECT_NAME}

(or longer) or similar? For example

iot_devices/site-temperature-management

I'm lazy to write a pull request, so I shall point out here that the link to the docs in

README.md

is broken: it links to

/core/

, which means accessing it from Github causes it to link to

<https://github.com/PrefectHQ/prefect/blob/master/core|https://github.com/PrefectHQ/prefect/blob/master/core>

instead of

<https://docs.prefect.io/core/|https://docs.prefect.io/core/>

I'd suggest linking directly to

<https://docs.prefect.io/|https://docs.prefect.io/>

as per the previous behaviour. @Anna Geller git blame says you made the change 😛

Hey everyone I am Amar. I am new to Prefect and am happy to be part of the community. Looking forward to learn from all of you. 👋

Hello everyone, tell me how it works, I do not understand. Is there an example available? here is the link

Hi there, I have this task with a trigger:

@task(trigger=any_failed)
def alert_email(email, message):
    sendmail(email, '[Prefect] ADCheck - [FAILED]', message)

with Flow("AD Check") as flow:

    # Parameters
    admin_email = Parameter("admin_email", default="<mailto:patrick@xyz.br|patrick@xyz.br>")

    # Tasks
    message = "some message"

    try:
        ad_conn = connect_active_directory(ad_server, ad_user, ad_pass)

    except Exception as e:
        message = f'Erro ao executar workflow: {e}'

    # If all tasks are successful
    notify_email(admin_email, message)

    # If any task fails
    alert_email(admin_email, message)

When the task fails, I would like to get the Exception error and save it o message variable, but I am getting

Trigger was "all_successful" but some of the upstream tasks failed

@Anna Geller, any tips? I've tried to use the state_handlers, but I need to be able to change parameters of the alert function for each workflow (email, subject, etc.) So triggers seems better. But I could not get the error.

Hi folks, I'm receiving this error everytime I try to run a Prefect script. For context I'm trying to receive Slack's status notifications so I added the SLACK_WEBHOOK_URL in the config.toml and I've also opened prefect's init.py out of curiosity. I don't think I changed anything in it but I might have miscliccked something.

File "C:\Users\andrea.nerla\AppData\Local\Programs\Python\Python39\lib\site-packages\box\box.py", line 516, in __getattr__

value = object.__getattribute__(self, item)

AttributeError: 'Config' object has no attribute 'datefmt'

My flow A has a StartFlowRun task that runs flow B. If I’m looking at a flow run of flow A in the UI, how can I navigate to the corresponding run of flow B?