Hello community, I just wanted to let you know that preview.prefect.io is not responding. If you are already aware of this issue, then ignore this message 😄 Wishes.

Hi- if you have some common code that should be shared between flows (let's say creation of some common resource, or a custom task type) what's the best way to lay this out? I'm currently putting this in a separate library and making sure it's available in the docker base image, but it feels a bit suboptimal as the only point of this code is to be used by the flows and so ideally it would live alongside them. Is there something better I can do here?

Hey I am a little confused on how to incorporate mapping to uploading to S3. I am using the first task to get all of the files. 2nd step to modify each file to a new json in memory. Finally, the 3rd task I need to upload those new files. Should I save the files locally before uploading? or is there a way to upload them in the same step of the mapping?

I have a question about the task decorator

@task
def get_file_name(table):
    file_name = f'bpi/{table}/-000'
    return file_name

with Flow("write to s3") as flow:
    exports = ['email_acquisitions', 'new_list_acquisitions', 'revenue', 'segmentation']
    for export in exports:
        file_name = get_file_name(export)
        print(file_name)

I was trying to write a function that does some string manipulation to get a s3 path (this is simplified ^) and when I add the @task decorator,

file_name

= <Task: get_file_name>, when I just want it to return the string produced. Am i doing something in a non-prefect way?

Hi there, has anyone done a performance analysis comparing Prefect to Airflow? Is there some sort of white paper I can use to present to my team?

When executing a flow locally, does using .map() with your tasks automatically does the parallelization using different workers(multiprocessing) or we need to use DaskExecutor for that?

I have a use case where I need to use multiple resource managers in a flow which cannot co-exist for reasons beyond my control. Is there a way to have a resource manager's initialization await an upstream task?

3,000 stars. Keep on rocking P

Why would this result in an error message in prefect cloud?

Why am I getting this error when I am trying to create flow run using the

client

object

prefect.utilities.exceptions.ClientError: [{'path': ['create_flow_run'], 'message': 'Flow 8eb08aff-f439-422f-94ba-31d30af635cb not found', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

Apologies if this has already been answered, but the docs don't make it super clear what terminology I should use so it's difficult to search for an answer. Basically, I want to know if we can have a flow where it waits for one task to complete, basically because some data isn't yet available from a provider, so it checks that for example every 15 mins and continues the flow when it's available? And importantly, is this available in the open-source version? This is the crucial reason we'd use Prefect instead of exclusively using something like

dbt

, which I think we'll use anyway.

Hello, I am looking into Kubernetes Tasks, and even though there is a

CreateNamespacedDeployment

task I am not sure, if it could start server based on docker image. Any suggestions?

what does the result ffolder holder in the .prefect folder?

sorry result folder in .prefect folder

It now has 516 gb of of HD space taken

can I delete some of the files here to free up disk space?

Hi! I'm trying to deploy the Prefect core server into an existing kubernetes cluster. Is there a recommended way to do this? I found docs on doing this for agents, but didn't see anything on doing it for the server (perhaps I missed it). My apologies if it's obvious, I'm a bit on the newer side to kubernetes in addition to learning about Prefect. If anyone could point me in the right direction I'd really appreciate it!

Hello....can you please share the firewall rule necessary to successfully run flow.register() ? We opened things wide in order to overcome the challenge and are looking to only open what is necessary. Thank you.

Hello! Can I test prefect cloud free trial behind company network?

Hi Everyone, I'm trying to run multiple dask-workers through Dockerfile and also dask-scheduler is it possible? The reason i'm doing this is to simulate a local run but using AWS EC2 instance resources

Hi! I am wanting to map a task, but I also don't want it to map concurrently in this specific instance. I want them to map one after another. Is there a way to do this?

hi folks, I’d like to keep some internal status across tasks in a flow run, every task will add some info into the status and the last task will process on the whole status. any suggestions to achieve that? thanks.

Hi! Is there a way to kill a flow run after a defined time? E.g. I want to kill a flow if it runs for 30 minutes, as my expected run time is around 10 minutes. This usually means something outside of my control is not working and I would just wait for the next scheduled run to happen. Any ideas? Thanks!

Hi!, Hope you all are doing fine in these harsh situations. Is there a way you can implement prefect with the class implementation? persay, with Flow "ok" as f" x =classA y= classB(x) Every time I create a flow in the above manner, I get a task runner object in "x" and classB doesn't obviously understand that. For now, am using this by implementing the same in a method, decorate task and run it. Is there a better implementation of this. any help is appreciated. Thank you. Stay Healthy.

Hi there, I added a parameter to a flow, but it still says on the dashboard:

@task
def task_process_data(date_to_process):
    if date_to_process is None:
        # calculate
        date_to_process = "some_value"


def flow_process_daily_data(*args, **kwargs):
    with Flow(*args, **kwargs) as flow:
        date_to_process = Parameter('date_to_process', default=None, kind=Parameter.POSITIONAL_OR_KEYWORD)
        task_process_daily_data(date_to_process)
    return flow

flow_process_daily_data(
            name="process-data",
            schedule=Schedule(clocks=[CronClock("0 14 */1 * *")]),
            storage=storage,
)
# Then I build the storage and register the flow here ..

This flow has no parameters ; Click "Run" to launch your flow!

Also, I don't understand why all the examples of Parameters have

flow.run()

called, as I don't call it, since it runs on schedule.

Hi everyone! I'm working on an internship project over this summer and have a question about working with Prefect states. Part of the goal of my workflow is to allow the user to manually trigger a task once a previous task has "failed". I plan on achieving this by implementing the

signals.FAIL()

state, which would then trigger another task. However, within that downstream task, I would also like to allow the user to interact with the flow to set a flag that would determine the outcome of that downstream flow, and this interaction would likely occur while the downstream task was in a paused state. However, I'm struggling with determining how I'd be able to achieve this interaction, as the paused state pauses all elements of the flow, thus preventing the user interaction I require. Additionally, if there are any other ways that a task can be set to occur both as a result of a failed previous task and manually at the same time, that insight would be greatly appreciated too. Thanks!

Hi everyone, I'm trying to setup a Fargate Dask client following the example from prefect in here with Dark Cloud Provider https://docs.prefect.io/orchestration/execution/dask_cloud_provider_environment.html#process. The problem is since the scheduler uses by default ports 8786 and 8787 for security reasons these ports are blocked at the AWS account I'm using, and because of that is causing the scheduler to time out. Is there any way to change these ports? Couldn't find anything in the docs just for Azure integration. Thanks!

Hi everyone, I'm new to Prefect and I am loving it so far. We will be using it for our production workflows to ingest external data. I have started to build my first dataflow using Prefect, but I have some questions. I feel like I need to demo it though to get some feedback. Is there a Prefect Support Group I can have a VC with to verify a few things? But the items I'm struggling with are: • Integrating GCP Stackdriver logging • It takes forever to push some data to GCS Bucket • Using GCP credentials

Hi folks - is there an example of using multiple script files when executing a flow (that are not part of a python package) ? I am using docker storage, and it seems that one solution is to specify the

files

dictionary and

env_vars

- the docker image is built successfully locally and passes the healthchecks, but when the k8s agent tries to create the prefect job I get

ModuleNotFoundErrors

Hi. I’m trying to implement this https://docs.prefect.io/orchestration/execution/storage_options.html#non-docker-storage-for-containerized-environments using minio storage and the Kubernetes agent. My simple flow runs but I’m getting this error. Does anyone know what can be causing this?

August 4th 2020 at 8:56:04pm | prefect.CloudTaskRunner
Unexpected error: NameError("name 'prefect' is not defined")
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/prefect/engine/task_runner.py", line 801, in get_task_run_state
    value = timeout_handler(
  File "/usr/local/lib/python3.8/site-packages/prefect/utilities/executors.py", line 188, in timeout_handler
    return fn(*args, **kwargs)
  File "minio.py", line 7, in hello_task
NameError: name 'prefect' is not defined