Is it possible to make it such that one task runs after all the other tasks are completed?

Hi everyone, we are building a flow where we are submitting steps to an EMR cluster. There are a lot of modules to be run. each module is a step. we submit the step and then check for its status. If the step is complete without errors we proceed to submit the next step. All of the modules/steps are in a huge list…….

[module_name1, module_name2,....]

. we are using a for loop to iterate over the list and submit the steps and check the status. However prefect does not submit these steps in sequence/order of the list. How do I make sure that the order is maintained ? Below is a screenshot of the code:

Hi, I’m trying to calculate some cost estimations for running cloud resources for our ETL pipelines in Prefect Cloud and to benchmark my own estimations I’d be curious to hear (just on a high level) what type of workloads your organization is dealing with and your monthly infrastructure costs. If you are using AWS EC2 instances (my plan is to use these for Docker Agent + DockerRun), I’d love to hear particularly the costs in this area in relation to your task workloads. I’m trying to find a cost efficient solution but want to start simple, although in future we may move more towards a setup described by @Anna Geller in this nice article: https://towardsdatascience.com/how-to-cut-your-aws-ecs-costs-with-fargate-spot-and-prefect-1a1ba5d2e2df Especially if scaling & maintaining the on-demand EC2 setup creates too much overhead

Just a quick one, we received an email with a summary of our throughput statistics and a champagne emoji - very nice, but I can imagine circumstances in which a client (not us on this occasion) may have to treat that as a security breach. I’m interested in opinions - anyone out there treating their number of tasks executed as confidential information?

Hi everyone, how do I name these tasks so that visually they appear with proper names ``````

I have a case where mapped runs seem to read the cache from other mapped runs. I'm using an approach liked the Complex Mapped Pipelines and have it set up to write a data pull process to a local result, cached for 10 hours, using

_result_=LocalResult(_location_='{plan}_plan_data.prefect')

. In my case my mappings are "plans". The problem though is that in one case a plan/map pulls data and caches it, and then a different plan/map subsequently reads that same data from cache and uses it!! Any idea what might cause this kind of behavior? It causes a bunch of my plan/maps to just not work.

Hi there, has anyone faced something like this before?

Unexpected error: TypeError('no default __reduce__ due to non-trivial __cinit__')
Traceback (most recent call last):
  File "/opt/venv/lib/python3.9/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/opt/venv/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 926, in get_task_run_state
    result = self.result.write(value, **formatting_kwargs)
  File "/opt/venv/lib/python3.9/site-packages/prefect/engine/results/gcs_result.py", line 75, in write
    binary_data = new.serializer.serialize(new.value)
  File "/opt/venv/lib/python3.9/site-packages/prefect/engine/serializers.py", line 73, in serialize
    return cloudpickle.dumps(value)
  File "/opt/venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py", line 73, in dumps
    cp.dump(obj)
  File "/opt/venv/lib/python3.9/site-packages/cloudpickle/cloudpickle_fast.py", line 602, in dump
    return Pickler.dump(self, obj)
  File "stringsource", line 2, in pymssql._mssql.MSSQLConnection.__reduce_cython__
TypeError: no default __reduce__ due to non-trivial __cinit__

I'm using Prefect 0.15.9 and pymssql 2.2.3. The task that raised it is the following

@task
def sql_server_get_connection(server: str, user: str, password: str, database: str):
    """
    Returns a connection to the SQL Server.
    """
    log(f"Connecting to SQL Server: {server}")
    # pylint: disable=E1101
    return pymssql.connect(
        server=server, user=user, password=password, database=database
    )

What would be the cleanest way to write a brief description about what a flow does in the UI?

Hi, trying to get only get the error logs from graphql. Does anybody know how I can restrict the query below so I don’t get all the logs levels?

query {
	flow_run{
    logs {
      message
      level
      info
      created
    }
  }
}

Of course a totally different query to get error logs from graphql would also be welcomed

What’s the difference/relationship between a flow group id and a version group id? I’ve noticed that as a republish a flow I see the flow ID change while the flow group id and version group id stay the same

Hi, how do i toggle local/cloud get of PrefectSecrets? I set a few keys in the cloud UI but I am unable to retrieve them with PrefectSecret currently

Is there a process to suggest a field get exposed in the graphql api? If so, I wasn’t able to see duration for flow runs as an available field, I understand it can be derived from start_date and end_date, but just wanted to put this out there

Hi - looking to pass AWS credentials to a Kubernetes agent at runtime so as to interface with the pynamoDB python library. I’m getting the following error when I attempt to scan a table in DynamoDB:

Failed to scan table: Unable to locate credentials

I have

AWS_CREDENTIALS

stored in Prefect cloud as a Secret, and I’ve tried passing credentials as envs passed to KuberenetesRun:

env={
        "ACCESS_KEY": Secret("AWS_CREDENTIALS").get().get("ACCESS_KEY"),
        "SECRET_ACCESS_KEY": Secret("AWS_CREDENTIALS").get().get("SECRET_ACCESS_KEY")
    }

Returns the same error. What’s the best / preferred approach to authenticate a Kubernetes agent against AWS?

What options are available for storing and running flows that span multiple files (e.g., flow.py, flow_tasks.py)?

Hello everyone, I am currently looking to try out a self hosted Prefect Server and UI instance on AWS. Based on the services I saw in the docker-compose file over here: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/cli/docker-compose.yml Has anyone attempted to deploy this on an ECS Fargate cluster ? Apart from the Postgres instance, I would like to deploy each service into its own Fargate instance. I am just wondering what could be some sane cpu and memory specs to kicks things off assuming I have a few flows with with an overall 1000ish tasks running twice every hour. I would also like to know if all these services are stateless, meaning I can scale them horizontally via a load balance ? Or are there any services here which can only be scaled vertically in case of performance bottlenecks ? Any help would be appreciated ! 🙏

Hey all, Im running a flow using VertexRun - It has multiple steps and it runs perfectly. Once a week at a specific time I’d like this flow to invoke another flow that will train a model. I used this code -

if pendulum.today('America/New_York').weekday() == 2:  # Monday is 0 so Wednesday is 2
        x_flow = create_flow_run(flow_name=PREFECT_TRAIN_FLOW_NAME, project_name=PREFECT_TRAIN_PROJECT_NAME)
        wait_for_flow_a = wait_for_flow_run(x_flow, raise_final_state=True)

This code is of course inside my “with Flow….” code block. Now when running this code alone (in a dummy flow) - it works and x_flow gets invoked. But when running this code in my real flow, after some other tasks - nothing happens. I cannot even see the task of “wait_for_flow” in prefect cloud(flow -> tasks tab) - seems like its getting ignored. What am I missing here? Thanks

Hello! Is there a way to run flows solely on EC2 agents without using ECS? I've seen some past messages about running flows on EC2 with ECS, but my team is looking to only use EC2 and spin up/down instances based on a custom image on our end.

Just want to say I have been focusing on getting prefect examples ready to show my team and after a few lessons learned it all worked like a charm. I am excited to be on the cusp with using Prefect to manage our data pipelines. The reporting and graphs already are way more than I expected. Thanks.

Hello, I have a situation where I am generating tasks at run time based on some configuration (so the number of tasks may be different depending on config). I would also like these tasks to be able to be run in parallel using the local dask executor. The difficulty I'm running into is that I need the tasks to be defined individually in the flow in order for the local dask executor to be able to run them in parallel. I do not know how many tasks there will be until runtime, so I cannot define the flow with individual tasks. I cannot simply loop through the task functions in the flow because the executor will not execute the tasks in parallel. Is there a recommended way to accomplish this?

Dear Prefect-Community, My Senior send me here to politely ask if it is possible to map prefect-task to individual kubernetes pods instead of paralell python-tasks. :)

I've been playing with Prefect Orion and I'm wondering how I can view the logs associated with a run in the UI?

Hi! How i can import my custom library to the flows run through kubernetes agent?

is there an equivalent to task.map but runs the mapped parameter elements in serial instead? basically identical to map but forced to run in order. I looked at the loop signal as an option, but I don't think that will lay out the schematic in the UI as nicely as map does

Hello all ! I'm looking for a simple way to modify a PrefectResult, in order to re-run a task with the correct input without re-running the whole flow. Is there a way to do it in Python, or otherwise with a mutation in GraphQL? I explored the schema but could not pinpoint where to find the result given its "location" id 🤔

Hi all! This may be more of a Docker SDK question, but was hoping a friendly prefecter might have encountered this issue. I'm trying to bind a Windows network path for my flow's run_config. I read the Docker SDK docs on create_host_config, but not sure what I am supposed to change. For example:   1) I can use the code below to successfully bind a local file path (for read/write) to a Docker container   2) However, if I use a UNC path to a network drive, I receive a "Server Error" that the path is not a valid Windows path Any help is welcome, thank you!

### THIS CODE ALLOWS ME TO BIND A LOCAL FILE PATH
from prefect.run_configs import DockerRun
import docker

  flow.run_config = DockerRun(labels=['my-label']
                            ,host_config={'mounts':[docker.types.Mount(target='/public'
                                                                        ,source=r'//c/some/local/path'
                                                                        ,type='bind')
                                                ]
                                        }
                        )

### THIS CODE THROWS ERROR 
  flow.run_config = DockerRun(labels=['my-label']
                            ,host_config={'mounts':[docker.types.Mount(target='/public'
                                                    ,source=r'\\path\to\windows\network\shared\drive'
                                                                        ,type='bind')
                                                ]
                                        }
                        )

👋 hey all, if there's anyone new to the community hanging out here, @Kalise Richmond and I are going to go live on Twitch in 5 minutes and go from 0 to first flow. See you at twitch.tv/prefectlive!

not wanting to disrupt the stream, do you guys have any timeline to prefect orion? looked at that and I'm wanting to start moving legacy code into prefect but I love how orion looks and didn't want to start moving and then having to deep dive again to change it once more ^^

How do I create and run flows on a schedule in Prefect Orion? Let’s say I have this:

schedule = IntervalSchedule(interval=datetime.timedelta(seconds=10))
deployment_spec = DeploymentSpec(name="hola", flow=greetings_flow, schedule=schedule)

Do I now use OrionClient and some methods to deploy? We currently use Prefect core in a pod and simply do flow.run() with a schedule attached, and I’m looking for a working example to do the equivalent in Orion (even if I gather behind the scenes it will do something else via a server and DB). Basically I’m looking for this: https://orion-docs.prefect.io/concepts/deployments/#running-deployments-with-the-api 🙂

Hi everyone, are results returned from the API sorted by

created

(or any other field) guaranteed to be returned in the same order? Thanks!