Hi everybody, did anybody using local agent notice any interesting spikes in memory usage? We use local agent and Dask executor. Our load is not that intensive at the moment - we run ~30 Flows during that spike. Regularly, we run into situation when the local agent's memory consumption spikes from ~ 150MB to 1.5-2.1GB. Has anybody had similar experience? At the moment, we're diving into debugging and profiling the issue so I'll post our findings here, if they turn out to be relevant.

Maybe it’s a stupid question but… How can I define the task name with the “_*create_flow_run*_” function?

Good Morning - I am experiencing an outage on my end w/ self hosted agents something is not allowing my agents to pickup flows - is anyone able to help me - are ports 8080 and 4200 the firewall rules I should check - not sure where to begin

Hi, at the top of my flow I'm doing sanity checking type stuff, if something isn't correct I

raise signals.FAIL('Didn't work')

I do not want the flow to proceed. However, other downstream tasks occur despite the exception being raised. Is this expected behavior?

Hi everybody! I have a flow that is scheduled to be run once every day. However, I would like that the executions during the weekend are moved to the next weekday (monday), which can be done with the adjustment function adjustments.next_weekday. However, I need the “real” date value (saturday or sunday) to run each flow correctly. Is there a way to use the original scheduled time before the adjustemnt? I don’t see any context variable for that. Thank you all in advance!

Hi - I have a flow that retrieves a list of keys from S3. Each key represents a .zip file which is an archive that when unzipped contains 5 different csv files. I am trying to map over those keys and write each file within the archive to Azure Storage. I am having issues understanding how to handle the zipped object - which at times should be unmapped. The working code I have right now only writes out the data associated with the last file in the list

Hi! I need little advice: how to update parameters using view. My default param is [{"1":"a"}, {}]. I want to add value [{"1":"a"}, {"2":"b"}] using view. I choose List in parameter type and put in my json. But prefect show me next value ["{\"1\":\"a\"}, {\"2\":\"b\"}"]. How to create list with objs using view?

hi, is it possible to add trigger in Prefect provided tasks? for example, I want to add a

always_run/all_finished

trigger to this task, but I can’t simply just add it to the end of parameters. Because this is not a

@task

way of definition, don’t know how to:

deletePod = kpod.DeleteNamespacedPod(kubernetes_api_key_secret=None)

hi all! I'm trying to use

create_flow_run.map(...)

to run a single flow multiple times with different parameters. I have that working, but unfortunately I need the created flows to run in a specific order. is there a way to force that behavior?

Hi! New to prefect and have a quick question regarding multiple flows. If i have a daily task and a weekly task that is the summary of the daily, should i start 2 local agents and register 2 flows? If the weekly runs every Monday for the previous sun-sat, should i set upstream even though it doesn’t require the same day’s daily task to finish running? thanks in advance.

Orion specific question, Why when I start orion agent

orion agent start

and I set the following env variable to connect to the orion server

PREFECT_ORION_API_HOST: orion_server
      PREFECT_ORION_API_PORT: 4200

the agent still try to create a database in this case using sqllite on the agent?

How can we see the agents connected to an orion server from the UI ?

Hi ! I’m wondering if there’s a way to run a flow every x minutes after completion of the previous flow run (as oppose to having a fixed scheduled interval) in prefect ?

Hi! I have 2 questions about flows versioning - 1. Is it possible to run archived version of a flow? 2. The documentation states:

You could override the automatic version promotion to unarchive and enable old versions (for example, for A/B testing)

How can I do that?

Is there a good example of having classes defined in a separate file and using that class in a task (on Kubernetes and Azure) - I keep getting the message module not found

Hi, Probably a noob question, but is there an equivalent to Airflow sensors in Prefect? In particular, suppose there is another team running an ETL that updates a core table in a data warehouse and I want my flow to start when once that table has been updated. In airflow I would use a sensor on that table. Is there something equivalent in prefect?

hi, is there a way to wait for a flow run to end other than polling

get_flow_run_info

and until the state is finished?

I'm looking for the way to get a flow to run with other classes, which are defined in a separate file. The idea being that we can encapsulate business logic in a class that multiple flows can refer to. We're not looking to create an installed package, just host the class file alongside the flow file. Is that able to be done? Just importing the class gives the flow error: "Failed to load and execute Flow's environment: FlowStorageError('An error occurred while unpickling the flow:\n ModuleNotFoundError("No module named \'doMath\'")\nThis may be due to a missing Python module in your current environment. Please ensure you have all required flow dependencies installed.')"

Hi All, Can we label Prefect agent after starting the agent? For e.g I have multiple test environments and I want to add another label to agent and use the same agent across.

Hi. I am refactoring. I divided my flows into 3 categories: • pipelines: Takes input from upstream pipe and feeds downstream pipe. Pipes are tube together with flow.update(pipe_file_reader) , flow.update(pipe_transform) and so on. • python parametered flows: Takes input such as pandas dataframe through 'Parameter' task. These are for paralelism with create_flow_run.map(flow_name) • production flows: End product flows These will be the sub folders of pipelines in my project. Any idea is appreciated!

hey, is it possible to data between flows using dependent flows?

def fetch_result(data):
    # data ?
    raise 

class GetData(Task):
    def run(self):
        return {'data': 7}

recipe_flow = StartFlowRun(flow_name="batch", project_name="proj", wait=True)                
with Flow(name='schedule') as sched:
    recipe_flow.set_upstream(GetData(), key='parameters')
    FunctionTask(fetch_result)(data=recipe_flow)

but the only thing I get as 

data

 is 

prefect.engine.signals.SUCCESS

object calling 

data.state

isn’t really helpfull, and 

data.state.result

 I get the 

prefect.engine.signals.SUCCESS

  all over again is there any way to get the real tasks/flowrunid/data of the the inner flow?

Is it possible to run a docker image with different environment variables in the container as specified by a Flow parameter? Locally, I can do this via

docker run -env FOO=BAR <my_docker_image>

I would like to be able to execute a flow on this image with a Docker Agent by calling the flow with the parameters

{
  "FOO": "BAR"
}

Is this possible with Prefect? Or is there a Prefect idiom for such a concept? Use cases for changing environment variables on container create/flow execution would be 1. defining which customer code path we want to trigger (database, configs, special methods and permissions) 2. Setting how we want to persist data (which database type to use, how to cache)

Do you know if there exists a pattern for varying the location of a result and hence the result type? My use-case is that on my laptop I don't want to re-run every task unless either the task definition has changed or the inputs have changed; instead I just want to use a cached result. When running in prod in prefect-cloud, when it's working again, I want to save to GCS but locally on the laptop, I'm fine with local storage. It would seem I can't set

@task(result=...)

because I then have to specify the result type to be either local or GCS. How do you handle this?

Hi all! I have a task that calls a SQL query and returns a list, and I want to iterate over that list and pass even chunks of the list to a scrapy spider that is called in another task and wait between each chunk. However, I'm running into an issue with LOOP where it is only passing the last chunk of the list to the scrapy spider.

@task
def query_that_will_return_a_list(): -> list

@task
def scrapy_api_call_chunks(title_list):
    loop_payload = prefect.context.get("task_loop_count", 0)
    title_list_grouper = list(grouper(title_list, 10))
    if loop_payload <= len(title_list_grouper):
        # Each loop will be an iteration of 10 titles. # of loops * 10 will result in the total number of titles looped over so far
        raise LOOP(message = 'Running the next 10 items in job titles list')
    scraper_class = Scraper()
    scraper_class.instantiate_web_scraper(title_list_grouper[loop_payload - 2])

I feel like I don't fully understand how to utilize LOOP in the context of passing information to another function inside the task.

Hey all! Was wondering if someone could help me out in configuring a prefect agent to make it a bit more robust. After we had an outage this evening, a massive backlog of flows occurred. This would be fine, except that the agent we have (a local agent run as a service in the docker compose) immedietly tries to launch all the delayed jobs, which makes it run out of memory. We get the Lazarus errors, the agent restarts, and same thing happens. You can see in the screenshot of service memory the constant agents restarting, using all available memory, and eventually crashing. Are there any settings/env variables I can set to try and a) limit the number of concurrent jobs the agent sends out and b) stop whatever is using memory? What exactly would cause an almost instant multiple GB memory usage in the agent?

Hey all! I am new to prefect, recently want to run a scheduled prefect job, but seems its being killed from time to time by a prefect ZombieKiller task, with an error No heartbeat detected from the remote task; marking the run as failed. Any idea why this happens, and how to make sure my task always have a heartbeat could be detected by the prefect ZombieKiller? thank you.

Question: Is it possible for 2 parent flows to access the same child run id?

Hey all! I am working on to integrate splunk with prefect agent. I need to know where all the LOGS being written, what is the path for the log files? Can anyone help?

Hey again 🙂 I’m trying to get the params that flow was run with in a state handler method but

new_state.result

keeps being an empty dict when running on prefect agent if i’m running the flow using

flow.run

I do see the results of the state

Quick question - if I have a set of tasks that don't depend on each other but that I want to run serially, what's the correct way to do that?