Hi Prefect Team, I have been using Airflow for some time. Heard about Prefect and gave it a try. Awesome product. I wanted some help on the following as it is an essential requirement for us :- • I have a ETL flow which runs every minute. It loads data from available in Kafka and processes it. • Sometime, the data in Kafka is very high and the flow continues beyond 1 minute. This leads to a pileup of scheduled flow runs. • Wanted to know if it is possible to not start the next scheduled flow run if the previous run of the flow is still running? I think, the Cloud concurrency feature is one solution. But unfortunately my system has very patchy internet(which is why data load fluctuates). So Cloud is not an option at present. Thanks.

Hi Im a Prefect newbie , Im try to start Prefect server and run my first flow , but when i run

/Users/hari/Library/Python/3.8/bin/prefect server start

im getting the below error

Docker version 20.10.8, build 3967b7d
Docker Compose version v2.0.0-rc.2

⠋ Container _postgres_1   Creating                                                                                      0.0s
Error response from daemon: Invalid container name (_postgres_1), only [a-zA-Z0-9][a-zA-Z0-9_.-] are allowed

the same works well if I run run using the docker-compose.yml based on the

/Users/hari/Library/Python/3.8/bin/prefect server config

Also can anyone help me with , where can I find help for deploying high availability community version

If high availability is important for your instance it is encouraged that you inquire about a community maintained deployment method

Getting the same issue as Hari ^

when i run docker compose on the created compose file

prefect server config > docker-compose.yaml

prefect server has errors but when i run

prefect server start

its fine. Do these two commands create the same docker compose file?

Hi Team, I see an issue in my Prefect server (0.13.10) environment. When the flow is removed during execution, there are a lot of errors reported and it will not stop. Is this an known issue?

prefect.utilities.exceptions.ClientError: Flow run ID not found: "3b51f006-cbbe-459c-b74f-6a832f22db96"

Hi there Since the version 0.15.5 i'm not able to register any flows to the server. I got this error:

C:\Users\hadornm\Miniconda3\envs\prefect-3.9\python.exe S:/cdwh-prefect/cdwh_flow_register.py
[2021-09-06 15:56:30+0200] INFO - prefect | Do DEV Register with image: xxx/xxx:1.x.x
Traceback (most recent call last):
  File "S:\cdwh-prefect\cdwh_flow_register.py", line 91, in <module>
    flow.register(project_name='cdwh ' + env_name, idempotency_key=flow_hash)
  File "C:\Users\hadornm\Miniconda3\envs\prefect-3.9\lib\site-packages\prefect\core\flow.py", line 1727, in register
    registered_flow = client.register(
  File "C:\Users\hadornm\Miniconda3\envs\prefect-3.9\lib\site-packages\prefect\client\client.py", line 1122, in register
    project = self.graphql(query_project).data.project  # type: ignore
  File "C:\Users\hadornm\Miniconda3\envs\prefect-3.9\lib\site-packages\prefect\client\client.py", line 548, in graphql
    result = <http://self.post|self.post>(
  File "C:\Users\hadornm\Miniconda3\envs\prefect-3.9\lib\site-packages\prefect\client\client.py", line 451, in post
    response = self._request(
  File "C:\Users\hadornm\Miniconda3\envs\prefect-3.9\lib\site-packages\prefect\client\client.py", line 737, in _request
    response = self._send_request(
  File "C:\Users\hadornm\Miniconda3\envs\prefect-3.9\lib\site-packages\prefect\client\client.py", line 647, in _send_request
    response.raise_for_status()
  File "C:\Users\hadornm\Miniconda3\envs\prefect-3.9\lib\site-packages\requests\models.py", line 943, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 405 Client Error: Not Allowed for url: <https://xxx.xx:443/>

Is this a known error?

I will not be able to expose prefect server to other machines on my network, have I understood that correctly? If start the backend on my computer, only agents on that computer will be able to register?

Hi everyone. I am having trouble with timing out flows. Some of them reach the set timeout but they ignore them. Now i have even set up an automation to cancel the flow on the UI, and it does cancel it, but it is stuck at "canceling". I have to manually set the state to canceled to free the slot. any clues?

Hope this is the right place to ask this question. This might not be conventional and without going in too much detail about why but I basically created a special instance of the flow like this which the

_run

creates a new key by default inside the context object so it can be available to every task that is added to the flow

class MyFlow(prefect.Flow):
    def _run(
        self,
        parameters: Dict[str, Any],
        runner_cls: type,
        run_on_schedule: bool,
        **kwargs: Any
    ) -> "prefect.engine.state.State":
        context = dict(my_variable="my_variable")
        return super()._run(
            parameters,
            runner_cls,
            run_on_schedule=run_on_schedule,
            context=context,
            **kwargs
        )

This works as intended when I call

.run

and the task is able to access that key but it doesn’t work if I register the flow with a local ui and agent and try running it from the UI. Why is that and how would I make it work if I need to register a flow and run it from the UI?

@task(name="log-my-variable")
def log_run_id():
    prefect.context.get("logger").info(f"{prefect.context.my_variable}")


with MyFlow("experimental-flow") as flow:
    result = log_run_id()


if __name__ == "__main__":
    # flow.run()
    flow.register(project_name="test-flows")

Brand new to Prefect. Are there any special considerations when running

prefect backend server

on an M1 mac? It seemed to have issues with hasura and graphql, so I killed the process after about 30 mins. Thanks

Hey there! After contributing the config option to schedule/run more than 10 runs per flow at the same time, we run into the next issue (we have been warned, I know 😬). When triggering more than ~25 runs, some of them (seems about 5 to 7) never complete (running over 24 hours now). Using LocalRun with a single LocalAgent. These runs are very lightweight: load a parameter, make a query with it and fail (intended). The states of the tasks are all either (trigger)Failed or Pending, so there is no actual work being done, it seems the state of the run just did not propagate. Clicking Cancel in the UI sets the state to Cancelling, it takes ages and the run is finally killed by the Zombiekiller. I'm going to try and randomize the schedules a bit to see if that relieves the issue. Do you have any ideas about where Prefect might lose track of these flow runs?

Hi, is there a way to have event driven flows executions?

Hello, I was wondering if there is a way to

inject

labels in a flow based on a project it is in. A bit of context, I want to run a flow on a Kubernetes cluster namespace based on the project it is part of (e.i each flow in a project

A

should run on namespace

A

) and I would like to use the agent affinity but don't enforce the user to specify those labels.

Hi all, is it possible to use prefect server cloud backend with a proxy in place?

Hi Prefecters, I'm getting some flows stuck in submitted state, perhaps one in ten or so. I can't see a solution here in the Slack. I'm running Prefect server in Kubernetes with a Kubernetes agent, more details in thread.

Hello 🙂 I’m experiencing a weird problem of prefect running the same task twice one after another in a span of several seconds. When looking at the schematic, it’s clearly seen that there’s only one task. I’m using prefect version 0.15.2. Thanks!

Hi everyone. I am trying to wrap the

Flow

class so that all tasks will report state to an external service. From what I undersand, subclassing the

Flow

class may not be a good idea as the prefect server will not use the subclass when managing the flows. I like that prefect has support for state change hooks (

state_handlers

), and I was wondering if I can use such a hook to modify the tasks of all tasks of a flow in order to add tasks state handlers that will report the task state change. Something like this:

def task_state_handler(task, old_state, new_state):
  report_state(task, old_state, new_state)

def flow_state_handler(flow, old_state, new_state):
  was_pending = old_state.is_queued() or old_state.is_scheduled() or old_state.is_pending() or old_state.is_submitted()
  if was_pending and new_state.is_running():
    for task in flow.tasks:
      task.state_handlers.append(task_state_handler)

Hi all! Maybe we have a weird use case, but I thought it should be possible with prefect : All our processes are dockerized, and we want to orchestrate them on AWS Fargate via an ECS Agent. We do not want to put the prefect flows on Fargate, but just instance tasks (e.g. running a container with a specific set of parameters). Now the conceptional question : what kind of task do I run? Do I need to also install prefect inside my docker image? Locally, I would just run my docker image via docker run or use the

docker.StartContainer

task. Thanks for your help!

What do I have to do to use GPU acceleration in a prefect flow? I'm using a DaskExecutor with cluster_class='dask_cuda.LocalCUDACluster' but I'm still getting "No cuda image available."

I saw some notes about the interactive ui enforcing pagination on graphql requests. Is pagination forced on all graphql requests?

Hello, I'm running into a weird error when trying to run a flow on a local agent. Whenever I start up a run I immediately get this error

Failed to load and execute Flow's environment: ModuleNotFoundError("No module named '/root/'")

Running in to issues deploying with Kubernetes using the default helm chart, enabling the tenant and the agent. The tenant and agent pods are having issues connecting. Agent logs ending with…

raise MaxRetryError(_pool, url, error or ResponseError(cause))
urllib3.exceptions.MaxRetryError: HTTPConnectionPool(host='<my-deployment-name>-apollo.<namespace>', port=4200): Max retries exceeded with url: /graphql (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7fb5d1bd6510>: Failed to establish a new connection: [Errno -2] Name or service not known'))

Tenant pod logs ending with…:

File "/usr/local/lib/python3.7/site-packages/requests/adapters.py", line 516, in send
    raise ConnectionError(e, request=request)
requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: / (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7fd0c274e990>: Failed to establish a new connection: [Errno 111] Connection refused'))

Not sure which settings need to be updated in the values.yaml file, but it’s gotta be something…

Has anybody ever run across

prefect server start

not working due to trying to create an invalid container name

_postgres_1

? I can’t figure out what changed, it worked last time I started the server a couple of days ago and now I can’t get it to work.

Is it possible to run local agent as a Windows Service?

Hi there, I set a prefect server on k8s and Minio as S3Storage. I created the docker image with the python dependencies. However, I’m having issue setting up the correct “AWS_CA_BUNDLE” as I’m using my companies certificates. I can see from prefect logs that it failed due to SSL verification. I sat up the correct env variable in the docker image, i.e. providing the CA file and setting up the “AWS_CA_BUNDLE” (see here: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#using-environment-variables). Any clue ?

Hi. I have an issue in a flow. It runs a task, that consists in runs several commands in a EC2 instance. Once it’s finished, it passed through a Manual Check, and then run another set of commands in the same EC2 Instance. The first step took about 1h 43 min, and once the final step is achieved, the status of the previous steps are being lost (instance_id, and commands to be executed). NOTE: InstanceId are the output of a task that creates the EC2 Instance. Do you have any clue of why task’s status could be loss somehow?

We are running our jobs on the Dask executor, in a pre-created Dask cluster. Are there any recommendations on how to scale the cluster with workloads or for added parallelism? (Ours run on Kubernetes).