Hi I have noticed that after restarting prefect server I have lost history and flows and I have to register all my flows once again. Is it possible preserve history and flows after stoping server?

Hi All, Prefect newbie here. I am currently evaluating Prefect for a switch over from Airflow. A particular usecase I am struggling with is accessing result of upstream tasks in state handler of current task. In Airflow, I could achieve the same by querying the XCom for the upstream task instance. Also is there a way to access task using its ID or name inside state handler? Any help would be really appreciated. Thanks.

Hi I am experiencing issue with

UI

as it not showing any

flows

and

runs

. I register flow with success and see in agent logs and server logs that flow runs when is scheduled but I have no information about flows and runs in

UI

.

My flows use KubernetesJobEnvironment and I specify a custom job_spec, but I'm noticing that I need to manually delete the k8s job between flow runs or else subsequent runs will fail to create the k8s job due to the job already existing.

Hi there, I'm trying to understand how the zombie killer decides to mark a task as failed and by extension how the heartbeats are actually sent. We use the local daemon with cloud which I believe forks a process per flow, who is doing the heartbeating there?

Hi, the following code crashes on me in Dask, due to

Large object of size 5.49 MB detected in task graph:

Am I doing something wrong? This is the simplest example I could come up with, that shows this behaviour.

Looking for best practices around continuous delivery around Flows. Specifically, we have a Flow that's triggered from another system over GraphQL. We're currently triggering the Flow on the

version_group_id

. The execution of the Flow includes a step to write data back to a Prod API of an external system. The challenge comes when making iterations to this Flow in development. The external system's Dev instance still triggers our Flow through GraphQL, but is expecting it to write data back to a Dev API. We could update and deploy the Flow to Cloud, but then the

version_group_id

picks up our Dev Flow, rather than our Prod Flow. The only thing I can think of is to have two different Flows deployed on Cloud (one for Prod and another for Dev). But in that case, promoting from Dev to Prod has a bunch of manual steps prone to human error. Interested in the communities thoughts on how you handle deploying multiple versions of the same Flow between Dev and Prod environments, where the configuration between Dev and Prod are different.

Has anyone managed to get Prefect installed into a Docker container for the purpose of building Flows into Docker? I keep running around in circles with the D-in-D problem, my brain is creaking…

hey everyone - weird question but does anyone have experience using Apache Spark. I’m investigating for a project best practices to run multiple parameter models on a single data set within a Spark cluster orchestrated with prefect. My main question is where to place the multiplier; would I get better performance to submit multiple job using

.map()

or submit a single job and manage running the different models within the single spark job.

Hi. I'm wondering what is the recommended way to run only parts of a flow. For example only tasks with certains tags.

Hey all, has anyone experienced issues where every so often the flow doesn’t execute because it fails to load custom modules? I’m using a local agent and have added the correct import paths. My flow was working just fine for the first few runs. Now, it runs maybe 1/3 times because it fails to load one of my custom modules. This could be a supervisord issue as well, but I’ve been doing some digging around and I’m unable to pinpoint where things are going wrong. any and all help is appreciated!

Hello 👋 can anyone help clarify my understanding how LOOPing and Context works between workers, especially (if it makes a difference) when using a distributed Dask cluster. Presumably the new task that is created is not necessarily executed on the same worker? And if not, can I safely rely on the newly LOOP’d task having the latest context? Here I’m specifically interested to ensure task_loop_result will reliably have the latest value but I am also generally interested to better understand how Context etc. is managed between workers. Hope that makes sense - feel free to point me at anything if this already explained elsewhere! Thanks! Andy

If you’re curious about Vue or front-end development, @nicholas is going to give @Laura Lorenz (she/her) an introduction to Vue in this Friday’s live stream — their goal will be to add a new tile the open-source UI from scratch, resulting in a new PR for Prefect Server! All experience levels welcome, you can sign up here: https://www.meetup.com/Prefect-Community/events/270547519

What's the best way to pass an object containing credentials? I'm making an ETL to get stuff out of Google Sheets, and I'm using the

gspread

package, which has you do everything from method calls to an authentication object. So like you go

gc = gspread.service_account(filename=<filename>)

and point it at a special credentials file, then everything is through that. Should I just pass it to Secrets? Is there any risk of sensitive credentials being cached somewhere if I define the

gc

object within the Flow itself?

Has anyone come across a case where you build a flow locally, and the register step is configured to register to a remote server (private prefect instance on aws, not prefect cloud). The output give me a url where the flow should be registered on that server, but there's nothing in the UI..

Hello! - Task question: is it possible to 1. re-run a task w/o triggering downstream dependencies? 2. do the same with ad-hoc task parameters?

Does prefect support micro-batching? In more detail: We’re trying to migrate our existing generator pattern to prefect - and we’re hoping we can change to a micro-batching model. Basically, our first node in our pipeline is responsible for pulling data from a location and pushing it out to the rest of the pipeline (a DAG). We were hoping to use the LOOP construct to have that “source node” pull data in

batch_size

increments, and map the individual data packets across the remaining DAG. In a way this kind of seems like a “workflow loop” with the parameters for the first node constantly updating.

Hi! Is there a regular release schedule for Prefect? I'd like to install a Prefect version that has the snowflake connector bump that's here: https://github.com/PrefectHQ/prefect/blob/master/setup.py#L56

For a given flow, if a task fails I am wanting to write the state to my local machine, and be able to load it back up later. I am looking into state handlers and persisting output using

LocalResult

but I'm wondering if there are Prefect features that are particularly appropriate for this use case.

Hi! Is there an easy way to make a task the last or first task of a flow explicitly?

Trying to understand why some flows die due to Zombie Killer when other, and then others, similar, don't. Sometimes see tasks succeed after 5 minutes or more without being killed, while others killed after 3 mins or so. With Zombie Killer enabled, does every @task wrapped function need to complete in < 2 minutes or risk being terminated? For example, if a task includes a subprocess.run() that does some long running work, is the heartbeat blocked until subprocess.run completes?

I'm on windows and prefect usually starts on restart just fine. However, I accidentally ran prefect server start again and now no flows show up on the UI. I even tired registering them again. Should i reinstall prefect? Flows look like they might be registering in the cli but they are not showing up on the ui. Prefect cloud seems to work fine though.

SO I registered my flows in prefect cloud. I tried to run one on demand. However, It is not running. It says running late. Am I missing a set up step here?

Question on testing and referring to task results. This example is contrived but maybe useful for discussion. Simplified class that defines a flow:

@task
def times_two(x):
    return x * 2

@task
def add(items):
    return sum(items)

class SimpleFlow(SRMPrefectFlow):
    @property
    def flow(self) -> Flow:
        with Flow("SimpleFlow", environment=env) as flow:
            x = Parameter("x", default=[1, 2, 3])

            times_two_task_result = times_two.map(x)
            flow_result = add(times_two_task_result)
        return flow

And a simple test to run the flow & check the last task's result:

def test_flow_run_result():
    flow = SimpleFlow().flow
    fr = flow.run()
    assert list(fr.result.values())[2].result == 12

The

list(fr.result.values())[2].result

works, but is fragile. We'd rather

fr.result[flow_result].result

but

flow_result

isn't available outside of the function that defines the flow. Is there a better approach that people have used?

Hi Is it possible to have dynamic

tags

? If I us map in one task and I want to limit operation besed on input to the task?

Hi, we're seeing some issues with Prefect cloud, none of our registered flows are available in the cloud UI as of about 30 minutes ago

Hello, Just to let you know that since

prefect 0.11.0

, you can't use the enviuronment variable

PREFECT__CLOUD__AGENT__LABELS=["hello"]

anymore but you have to wrap the values in single quote like

PREFECT__CLOUD__AGENT__LABELS='["hello"]'

I know it is documented with the right format in the doc but it was working with a "bad" format before and so it can broke your flow after upgrading to v0.11.0 👍

I'm getting a weird error from the apollo service whenever I try and run a flow. The error I see in the logs is this

apollo_1     | 2020-05-15T16:42:41.530Z {"message":"Cannot query field \"setFlowRunStates\" on type \"Mutation\". Did you mean \"set_flow_run_states\" or \"set_task_run_states\"?","locations":[{"line":2,"column":5}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}}

apollo_1     | 2020-05-15T16:42:41.547Z {"message":"Unknown type \"writeRunLogsInput\". Did you mean \"write_run_logs_input\", \"write_run_log_input\", \"create_flow_input\", or \"archive_flow_input\"?","locations":[{"line":1,"column":18}],"extensions":{"code":"GRAPHQL_VALIDATION_FAILED"}}

Not sure how I go about debugging this, any tips?

Hi all! @nicholas and I are queued up for a livestream today on how to contribute to Prefect Core’s open source UI in 15 minutes! Feel free to drop by, or go to this link after for the recording if you can’t make it!