Hello, how can I reuse a flow? I have a function that takes in a list of items and build a flow using those items (I don't use prefect parameters currently). I want to deploy 2 versions of that flow (they will be different since the item list will be different) . I do change the name each time but it seems like they're both being considered as the same flow and replace each other.

Hey all, happy Tuesday!   I'm trying to import additional generated Python files and use them as part of my Prefect Workflow. Everything works fine locally, but when I push up the flow to a docker repo and try to run it using a Kubernetes agent, I'm getting the following error from cloudpickle:   ModuleNotFoundError: No module named 'test_pb2'   The Kubernetes agent spins up a job in our cluster, and this error comes from that job. It works fine if the flow doesn't reference any other custom modules.   Here's what I've tried so far, but it all results in the same error (note that the healthcheck.py passes just fine in all these cases when the flow is pushed to docker storage):   1.       Setting PYTHONPATH to point to test_pb2 and other files. If I pull the container, PYTHONPATH is correct; however, when the kubernetes agent spins up a job using that container, it doesn't appear like it's set anymore. Is it getting overridden somewhere?   2.       Turning the generated files into a package installable with pip and then installing it as part of the Dockerfile that creates the docker storage. If I pull the container, the package appears to be installed successfully. Still the same error when run by the kubernetes agent though.   Any ideas on other things I could try? Is there a preferred way to include other Python files as part of a prefect flow?   Any ideas/help is greatly appreciated!!

Hi All,

is there a way to make the prefect logs in the ui display in ms or does this even log at this granularity?

Hey team, recently some users have been asking us to announce core releases here on a regular cadence so we are going to start doing that again! Just released

0.13.3

and here are a few notable changes: 🌲  More control over log levels 📦  Pass any arguments to Docker tasks 🔑  Fixed caching when no cache_key is present A big thank you to our contributors who helped out with this release! Full changelog:

Is there right way to check executions on dask cluster? My flow:

from prefect import task, Flow, case, Task
from random import randint
from prefect.schedules import IntervalSchedule
from datetime import timedelta
from prefect.engine.executors import DaskExecutor
from prefect.environments import LocalEnvironment


class Preprocess:
    def __init__(self, number):
        print(number)
        self.number = number
        self.data = None

    def read_file(self):
        self.data = self.number

    def preprocess_file(self):
        self.data = self.data * 2

    def save_file(self):
        self.data = self.data / 3


@task
def is_running():
    return False

@task
def get_files():
    return [i for i in range(0, randint(0, 15))]

@task
def _print(lst):
    print(lst)
    print(len(lst))

@task
def etl(file):
    prep = Preprocess(file)
    prep.read_file()
    prep.preprocess_file()
    prep.save_file()


executor = DaskExecutor(address="<tcp://localhost:8786>")
schedule = IntervalSchedule(interval=timedelta(seconds=5))
local = LocalEnvironment(executor=executor)

with Flow("process", schedule=schedule, environment=local) as flow:
    condition = is_running()
    with case(condition, False) as cond:
        files = get_files()
        _print(files)
        etl.map(files)

flow.run()

Accorging tuto : https://docs.prefect.io/core/advanced_tutorials/dask-cluster.html I started scheduler and 2 workers , and after looked at the boker-web ui. But Tasks are empy on ui.

Hi all! Is this an okay place to ask getting-started questions?

I'm trying to get started with the kubernetes agent locally (and just

prefect server start

for the scheduler) it keeps firing up pods that try to connect to localhost:4200 for the graphql endpoint and failing because they can't access

localhost

from inside a pod. Is there a way to specify the endpoint the pods should be trying to connect to (e.g. to `kubernetes.docker.internal`instead)?

Tried reading through the docs and the source but I didn't see anything, so I figured I must be trying to do something non-standard

Hi all. Using Prefect Cloud i've setup a user with read-only role, but they can't see any projects. Doesn't read-only allow them to see everything in the cloud? If not how to apply those permissions?

I am having an issue where I created an account using google and then was sent an invitation to my organization's (Datateer) prefect. I followed the invitation and in messages it says that I have joined the group, but I can't see any of the organization's materials. Thanks for the help!

Hi all, does prefect provide the feature like ""depends_on_past=True"" in Airflow when running flows on schedule? Thanks for the help!!

Hi all - we’re using Github Flow Storage and from time to time our flows fail with:

Failed to load and execute Flow's environment: ConnectionError(MaxRetryError("HTTPSConnectionPool(host='<http://api.github.com|api.github.com>', port=443): Max retries exceeded with url: /repos/<ORG>/<REPO> (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x7ff5caf88668>: Failed to establish a new connection: [Errno -2] Name or service not known',))",),)

What are potential causes for this error? When there is downtime on Github, we see

Failed to load and execute Flow's environment: GithubException(500, None)

. What part of the source code should I explore to understand this better?

Hi everyone! I have a problem with event driven/streaming flow.

Hi. I'd really appreciate some help on connecting to a remote k8s cluster running in azure. I can execute flows in the cluster when using

flow.run(...)

but when I

flow.register(...

) and then run it from the UI, I get errors in the dask workers in k8s saying that cannot connect to

localhost:4200

. which makes sense since that localhost:4200 has to be replaced by the IP where the prefect server is running. how to do that? (https://github.com/PrefectHQ/prefect/issues/3185)

We use the prefect logging method. Either self.logger from

Task

or imported from

from prefect.utilities.logging import get_logger

We get some weird double logging with different formats. I am not sure if it is a pure Prefect issue or just something we do wrong on our ends. But did someone else see something like this an has a solution?

[2020-08-19 15:21:50] INFO - prefect.TaskRunner | Task 'load': finished task run for task with final state: 'Success'
Task 'load': finished task run for task with final state: 'Success'
[2020-08-19 15:21:51] INFO - prefect.TaskRunner | Task 'extract': finished task run for task with final state: 'Success'
Task 'extract': finished task run for task with final state: 'Success'
[2020-08-19 15:21:51] INFO - prefect.FlowRunner | Flow run SUCCESS: all reference tasks succeeded
Flow run SUCCESS: all reference tasks succeeded
[2020-08-19 15:21:52] INFO - prefect.orchestrator.utils | Flow state transition: <Running: "Running flow."> -> <Success: "All reference tasks succeeded.">
Flow state transition: <Running: "Running flow."> -> <Success: "All reference tasks succeeded.">

Hey guys, how would you change

BigQueryTask()

arguments at runtime? Basically I would like to be able to change the

table_dest

argument based on an environment variable, so that I can save my query result in a different place depending on if I run the flow locally or elsewhere. Thanks very much in advance

Hey Team, Does anyone have the issue of not being able to restart failed tasks? It seems that it should be okay, asking for a click confirm, but it is often grayed out.

Hey all. Does anyone know off hand why a flow wouldn’t get picked up after it’s been submitted? The last version of it got picked up. I’ve deleted the flow, bumped the version tag, rebuilt, and still nothing. Get lazarus killed 3 times and errors out. Agent is running other flows fine.

Hi, is there any equivalent in prefect to Airflow's automatic picking up of new dags? Or do all flows need to be explicitly registered?

Sorry for the battery of questions- the KubernetesJobEnvironment overwrites the image I provide in a job spec with some flow specific image-- how can I specify a flow should run with a specific image?

good evening! Are there any plans to include a role-based system in the server release? or will it only stay in the cloud?

Question: if I have a task I want to perform over all the lines in a file, but have each task run on a line depend on the next (I don't want to continue iterating if a previous line fails), whats the best prefect way to do this? for right now I'm doing:

with Flow('file') as flow: 
    name = Parameter('name')    
    lines = read_file(name=name)
    for l in lines:
        do_line(l)
    finish()

I know theres a prefect

Map

functionality, but I didn't see an easy way to make it so each task for each line depends on its previous line task in an iterative manner

Hi All, I have a flow that I wrote, and I started a prefect service using:

prefect backend server &
prefect agent start &
prefect server start &

Then I registered my service using:

flow.register(project_name="ML")

Finally went to the Web UI and click on quick run, but then it’s sort of been stuck there for a while now Is there any way to debug why this is happening?

Hello, I wanted to check if someone can suggest community maintained deployment method to deploy prefect on multiple Windows servers? And maintaining one database for these deployments.

Does anyone else in the community use type-hints to track data-dependencies in a Flow? Creating tasks with the

@flow

decorator works well for me, but the class-based API confuses my IDE. For example, my IDE thinks the class-based result should be type `Task`:

from prefect import Flow, Task, task


@task
def func_task() -> str:
    return ""


class ClassTask(Task):
    def run(self) -> str:
        return ""

class_task = ClassTask()


with Flow("test_type_hints") as flow:
    func_result: str = func_task()
    class_result: str = class_task()  # Expected type 'str', got 'Task' instead 
flow.run()

When I use

flow.run()

everything works fine, but when I try to use the prefect server, it’s complaining about environmental variables not setup for AWS, such as:

KeyError: 'No access key found. Please set the environment variable AWS_ACCESS_KEY_ID.'

Is this because the runs are being run inside the docker container?

Hi all, seems the local agent execute a flow run in a separate process, I’d like to do some initializations at the very beginning of flow run, maybe right after the process is created . Is there any way to achieve that?

Thanks