Hi all, newcomer to prefect - happy so far. I'm stuck a bit in a Dask integrated prefect environment. I have a simple flow that looks something like this:

@task(log_stdout=True)
def say_hello(name):
    print(f'{datetime.now()}: workflow hello {name}', flush=True)
    worker = get_worker()
    return f'done on {worker.name}, scheduler at {worker.scheduler.address}'

name = Parameter('name')
with Flow("Simple parallel hello") as flow:
    # Since there is no return value dependency, we end up with possible parallel operations
    for i in range(10):
        say_hello(name)

If I run the flow from my script, targeting a particular Dask cluster, I can hit the right Dask workers:

executor = DaskExecutor(address=dask_scheduler_address)
flow.run(executor=executor)

My question is about registering this flow and running it, say, from the prefect ui. I can easily register the flow with:

flow.register()

But then trying to run it from the ui just hangs. I'm pretty sure it is because the executor isn't registered with the flow. Am I missing something? Thanks in advance...

Hello! I have an issue in production, graphql seems to be down : when I call xxx.eu-west-2.compute.amazonaws.com:4200/graphql I have 4200: Connection refused and when I check docker container ls it's not on the list - do you know how I can restart it ? PostGre is also down

Hello everyone, anyone have an idea on how to write a GQL query for marking as success upstream tasks? What i'm trying to do here is to run a dag from a specific task forward. I had bug in my code, fixed id, registered a new version, and now I have to rerun the dag. The thing is that alot already succesful tasks would ran again It takes almost one hour to run again.

question about mapping tasks -is it possible for a task to be 'mapped' over multiple iterators? (think task with 2 parameters and I want to 'map' over both with a list of param values)

Hi all, I'm working on some custom prefect serializers for pandas df results. I have something that seems to work but I don't think it is very robust (requires creating a local file).

class ParquetSerializer(Serializer):

    def serialize(self, value: pd.DataFrame) -> bytes:
        # transform a Python object into bytes
        tmp_filename = str(time_ns()) + '.parquet'
        value.to_parquet(
            path=tmp_filename,
            index=False
        )
        with open(tmp_filename, 'rb') as in_file:
            df_bytes = in_file.read()
        Path(tmp_filename).unlink()
        return df_bytes

    def deserialize(self, value:bytes) -> pd.DataFrame:
        # recover a Python object from bytes        
        df_bytes_io = BytesIO(value)
        df = pd.read_parquet(df_bytes_io)
        return df

Does anyone have thoughts about the above approach? (saving as a local file then reading the bytes from the file?)

Hi again, I wanted to mention that I've successfully replaced airflow with prefect in production for all our ETL/BI and ML workflows at Instasize.com! Big Thanks for all the help/suport along the way. We ended up building a google cloud centric solution around prefect core. We did look into using prefect cloud (I like the dash/features, cost not an issue) but had friction with the implementation at this time. If anyone on the team is interested I'm happy provide any feedback about our experience. Thanks again!

When I register a flow, I'd like to save arbitrary metadata with the flow version using the Prefect Python API. Can anyone suggest a way to do so? I would like to look at the flow version in Prefect Cloud and perhaps see my custom key/values in the details.

Hey guys, I am looking for a solution to abstract computations from supercomputers, mostly for fmri data preprocessing (and, on a lower note, analysis on the underlying output) using already developed binaries within singularity (docker-like) containers on large static dataset. Each fmri session is to be processed parallely. We need to assess the current state of the computing environment (is the inode / lustre file limit close to be exceed, etc) and adapt/restrict the job count to prevent crash + perhaps remotely archive outputs and cleanup space as it is going. This can technically be done beforehand using some heuristic or dynamically. Would you consider prefect (relying on stg like dask_jobqueue.SLURMCluster) to be a good fit to develop such system, or would you have something else to recommend ?

(On top of that, we’ll need some versioning system + manual quality control of the data through a web interface.)

Hey @josh, one more PR to add

extra_docker_kwargs

to

CreateContainer

https://github.com/PrefectHQ/prefect/pull/2915

I guess we can’t run prefect with singularity instead of docker (cf. https://docs.prefect.io/core/getting_started/installation.html#running-the-local-server-and-ui). Would this be stg easy to change/patch for an external contributor ?

Hello everyone, I have a couple of questions regarding deployment. I am considering using Docker or GCS Storage for my flows and trying to understand what makes the most sense. 1. When I use either, do I still run the agent in a virtual environment that includes the imports used by my tasks? Perhaps one is different than the other in that aspect? 2. What is the usual way to register a flow? Do people just use an ad hoc script whenever they have a new flow or perhaps do some kind of diffing & registering what isn't there as part of a release process? When is re-registering a flow ever required? Perhaps when changing the schedule?

Hi, can anybody help me with the docker-tasks? I want to design a flow to renew an docker image: 1. pull the latest MQTT-image (docker pull eclipse-mosquitto), 2. remove the running MQTT-container (docker rm MQTT-NAME) 3. start the new MQTT-container (docker run -tid -p 1883:1883 -p 9001:9001 --name=msg-broker --restart=always --network %x_Network% eclipse-mosquitto) How do I phrase that in Prefect? I can't even pull the latest image?! Thanks a lot!

I’m running into an error using

flow.replace

, which seems to have to do with the fact that

flow.remove(task)

doesn’t remove

task

from `flow.slugs`:

from prefect import Parameter, Flow, Task

p1 = Parameter('p')
p2 = Parameter('p')
t1 = Task()
t2 = Task()

flow = Flow('flow')
flow.add_task(p1)
flow.add_task(t1)

print(flow.slugs)
# {<Parameter: p>: 'p', <Task: Task>: 'Task-1'}

flow.replace(t1, t2)
print(flow.slugs)
# {<Parameter: p>: 'p', <Task: Task>: 'Task-1', <Task: Task>: 'Task-1'}

flow.replace(p1, p2)
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-7-5ec138031eb6> in <module>
----> 1 flow.replace(p1, p2)

/opt/conda/envs/drugdiscovery/lib/python3.6/site-packages/prefect/core/flow.py in replace(self, old, new, validate)
    291         # update tasks
    292         self.tasks.remove(old)
--> 293         self.add_task(new)
    294 
    295         self._cache.clear()

/opt/conda/envs/drugdiscovery/lib/python3.6/site-packages/prefect/core/flow.py in add_task(self, task)
    482                 raise ValueError(
    483                     'A task with the slug "{}" already exists in this '
--> 484                     "flow.".format(task.slug)
    485                 )
    486             self.slugs[task] = task.slug or self._generate_task_slug(task)

ValueError: A task with the slug "p" already exists in this flow.

Hi folks, what is the simplest way in prefect to specify that a task only execute based on the value of a boolean parameter, otherwise should get skipped ? nevermind - I just spotted signals

from prefect.engine import signals

@task
def signal_task(run_task: bool):
   if not run_task:
       raise signals.SKIP()

Hello from Chicago! I've tried searching this Slack and the Prefect documentation but haven't found a good answer for this.

How should I handle prefect version differences between the environment where a flow is created and registered (

flow.register()

) and the environment where the agent runs?

So, concretely, let's say that I have an agent running using the image

prefecthq/prefect:0.12.1-python3.7

. I have some flows already being run by that agent, and those flows were created with

prefect

0.12.1. Imagine that

prefect

0.13.0 has just been released on PyPi, and data scientists on my team are going to

pip install prefect

, create flows, and register them with

flow.register()

What should I do? 1. Add a label to each agent with the prefect major + minor version (e.g. v0.12, v0.13). Make sure flows are registered with such a version label, so 0.12 flows run on the 0.12 agent and 0.13 flows run on the 0.13 agent.

prefect

uses semantic versioning, which means that there can be breaking changes between minor releases in the

0.x

series. 2. Do nothing. a flow created with 0.13.x should be expected to work with an agent running 0.12.x 3. Upgrade all those 0.12 flows to 0.13. Prefect Cloud is only ever running a single version of

prefect

and if your flows + agent are not in sync with that version, bad things will happen. 4. something else Thanks for your time and consideration!

Hi all, I'm struggling to get the 'zip' mapping pattern to work, I'm probably missing something basic here. I'm tying to map a task with 2 parameters. I'm looking at the reply to this github issue (https://github.com/PrefectHQ/prefect/issues/1986) where it is suggested to use the

itertools.product

function to map over multiple tasks. However, I can't get this to work in a simple example:

import itertools
from prefect import Flow, task
from prefect.engine.executors import LocalDaskExecutor


@task(checkpoint=False)
def cross_product(x, y) -> list:
    return list(itertools.product(x, y))


@task(checkpoint=False)
def concat(a:str, b:str) -> str:
    return a + b


a = ['d', 'o', 'g']
b = ['c', 'a', 't']

with Flow(name='zip-map-test') as flow:
    
    result = concat.map(cross_product(a, b))


executor = LocalDaskExecutor(scheduler='threads')

flow_state = flow.run(executor)

Could someone help explain how EnvVarSecrets work with Docker storage? I'm registering my flow against a local prefect server and have a docker agent scheduling it successfully, but I can't seem to get environment variables to be in scope no matter what I do. I've tried setting the env variables in the shell that runs

prefect server start

, the one that runs

prefect agent docker

, and the one that runs the script that registers the flow, all with no luck

Hello everyone, I'm looking to try out prefect for one of our pipelines. We have multiple "sources" that consist of 2 "highlevel" types: base and parent sources. Parent sources have a list of children sources. We may choose to add or remove sources based on a config. Each source has a

run

method, calling run on a parent source calls the

run

method for each of the children. Currently, all the top-level sources are called sequentially and then finally aggregated, does anyone have any ideas on the most "prefectic" way to structure this to maximize parallelization and conciseness? One way is to define a task for each highlevel source, but this seems to be a bit tedious. We also won't be able to parallelize the children. Another approach I tried is something like this.

@task
def run_for_child(child):
    child.run(max_items=50)
    return child.collection_name

@task
def run_source(source):
    cond = source.is_parent
    with case(cond, True):
        res1 = run_for_child.map(source.children)
    with case(cond, False):
        source.run(max_items=50)
        res2 = source.collection_name
    return merge(res1, res2)

@task 
def get_highlevel_sources(config):
    # return list[Source] based off config

# In the flow, get the sources and do runsource.map. At the end, aggregate all collections

This gave me a

could not infer active flow context

and I'm not sure if this is the best way to structure this anyways

$1,000,000 idea for y'all free of charge: what if the prefect favicon for a flow run page showed the color of your flow run status?? like GitHub and CircleCI do

Hey community, I’m trying to figure out a way to control the flow with the caches. Suppose I have a flow with a dependency graph like this: A_heavy_mapped_task --> B_heavy_reduce_task --> C_some_other_task Now, suppose I have the results for tasks A & B cached in a

Result

and I just want to run task C. In Luigi, C would ask for B’s output, and only fetch that. However, in prefect, the flow runner will fetch ALL the results of the mapped tasks A, but since we already have B’s result, that’s totally redundant and wastes a lot of time and data exchange. Any idea on how to tackle this issue? i.e. fetch only task B results somehow so that C can run?

Hey guys, I am new to the community and was trying to use prefect with docker. My use case being, I want to use the prefect server along with dask and run the same on my own ec2 instance using docker. I wanted to ask how to spin up the docker container for the same and when I have to register new flows to the server, I would want it to be as simple as adding the script to the src folder. I am a bit confused here and would appreciate the help. TIA

Hello everyone! I'm using Prefect and everything works well, but.. I want to secure the prefect UI with username and password, anyone can help me with this? TIA 🙂

Has anyone else run into an issue with duplicate tasks being created? I have a list of Task instances that I would like to run in parallel.

with Flow("My flow", schedule=schedule) as flow:
        for source in all_sources:
            source.bind(
                ## parameters for my run() function here
            )
        res = [source() for feed in all_sources]

In the schematic and gantt chart I have a dag with all the

sources

leading to a list, which is expected but I also duplicated

source

tasks without any edges

Hi - Is there a way to check if a list is empty in the positive?

with Flow("hello"):
    some_lst = make_lst()
    with case(some_lst.not_(), False):
        # do something

I am trying to figure out the best way to have a conditional branch in my flow, probably involving the

ifelse

fn in prefect. I have a task, we can call it Task A, that queries my database for a value and returns it. If Task A returns some value, then that value should get passed to Task B, whose output gets passed to Task C, whose output goes to Task D. But if Task A returns

None

, then I don't want Task B/C/D to run. Right now I have it set up like this:

TASK_A = query_database_task()
RUN_TASKS_B_C_d = dummy_task() # task does nothing
ifelse()is_null(TASK_A), RUN_TASKS_B_C_d, None)

TASK_B = some_task_b(TASK_A)
TASK_B.set_upstream(RUN_TASKS_B_C_d)

TASK_C = some_task_b(TASK_B)
TASK_C.set_upstream(RUN_TASKS_B_C_d)

TASK_D = some_task_b(TASK_C)
TASK_D.set_upstream(RUN_TASKS_B_C_d)

There has to be a better way to do this

question about results targets cacheing, does prefect read the entire cached file or just check if it exists before deciding to run vs. skip?

quick q on mapped task inputs (@Jim Crist-Harif mentioned here that this might that this might get resolved in the mapping refactor but we're still seeing it) is passing

task.map(sequence, x=unmapped(some_large_result))

a bad idea? we're seeing

UserWarning: Large object of size 874.55 MB detected in task graph

and very very slow creation of mapped tasks; I would have expected that these would be re-using the dask futures already present on the cluster, not rehydrating the outputs inside the

FlowRunner

loop, but I guess that's not how it works. is this expected behavior and if so is there any known workaround..?

Hi all, I was wondering how to make the parent flow wait for the FlowRunTask child flow output. What I mean here is that I don't want the FlowRunTask to only trigger the second flow but I want to be able to wait for it to end and use its output in the parent flow. Thanks in advance!

Hey guys. I have a question regarding prefect 0.12.2 and the slack_notifier. We are running on prefect core without prefect cloud. By default, the slack notifier tries to connect to the prerect cloud server to get some api token in slack_message_formatter in notifications.py. looks like if there is a prefect.context.flow_run_id and prefect.config.backend is set to cloud, try to connect to cloud. I have tried to both set flow_run_id to None and also backend to "cloud" but that only had an effect when the flow itself tried to connect to slack. It did not had an effect on individual tasks. Is there any propper way of configuring this, other than a local prefect/backend.toml?