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

Hi guys! I found a weird behaviour with the agents: We have a flow that has been working perfectly for a while now. The problem is that every now and then we have a run in which every task (in the same flow run) is executed twice. We have only two agents running in an EC2 instance. I started looking today and I found that sometimes both of the agents pick the same flow run, and both execute it! [more info in the comments so I don't pollute the main thread]

Hello! I'm having some trouble with running a flow on a single node dask cluster on Kubernetes. When I run the flow locally, the flow works as expected, but when I run it on Kubernetes , I get the following error:

Unexpected error: TypeError("Cannot map over unsubscriptable object of type <class 'NoneType'>: None...")

. The problems seems to come from a mapped task that outputs a pandas dataframe. Anyone has any ideas on how to fix this problem or how to debug it further?

Good morning, whats the easiest way start a flow from python that is already registered?