Hi @Kevin Kho, Orion is getting more attractive now. But the collection is only maintained by prefect, can I contribute to Orion task collection? Or any guide to create my own task collection and share that to other developers?

Hello There My name is Himanshu Pandey and I am a security researcher. I have read privacy policy and I have found vulnerability in prefect services. Please contact me ASAP. I hope you understand me. Thank you, Regards.

Hello, I am evaluating whether to use Airflow or Prefect for a side project that includes extracting data from a API and loading them in Snowflake. My understanding is if I use Airflow, I need to use the cloud vendor such as AWS MWAA, which would incur some cost. In comparison, using the Prefect cloud to start out is free. Is my understanding correct? Thank you!

I seem to be having a problem passing a

Parameter

into the class constructor

__init__

of a task that has been subclassed from

prefect.Task

I am receiving a

prefect.Task

object instead of the value from the Parameter itself. I can't seem to figure out what I'm doing wrong.

Hello, we are evaluating prefect and have a particular use case in mind that requires us to (1) Run high level workflow steps in "Project A" (2) Run specific (smaller) steps of the above workflow in "Project B" Is is possible to deploy an Agent to each project so that Prefect is managing the entire execution (we currently perform cross-project RPC calls but would like to move away from this approach)

Hi, I am wondering where the logs and run histories are saved. Can anyone point me to the right direction? (Im using prefect-server as backend)

Hi all, my company is testing prefect and I am having trouble installing a kubernetes agent, which I suspect is because of the company proxy. When I try to install the agent the logs say that it is being registered, but then eventually I get an SSL error.

requests.exceptions.SSLError: HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Max retries exceeded with url: / (Caused by SSLError(SSLCertVerificationError(1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1091)')))

Is there a way to provide proxy credentials when starting the agent? I've tried using the --env flag to supply proxy information. Or is there something similar to a --verify=false that I can set?

Hi Team, We are using prefect core as of now and thinking about switching to prefect cloud. Can i get a link or documentation details around what are key advantages of cloud offering vs core product. Also, compared to core product, how much we can still control in prefect cloud and which specific things will be out of end users hand and will be fully managed by cloud offering ?

Hey, we are thinking about developping a simple mvp web app that needs to pulls data from an external api and do a basic visualization. Do you think we can do it with prefect as the EL tool ? Even tho there is no real Load part as it will only consist of eventually parse/transform the response of the external API and directly pass it to the front. I was thinking about triggering the EL flow with the front end calling the Prefect API and the response would be the data collected by the flow. Is it possible or do you recommend another way ?

We've run into another issue that I was wondering if anyone could help clarify. I am running a dask cluster on an on-premise kubernetes cluster and I have the kubernetes agent running in the same namespace. The agent is connected to prefect cloud and the jobs are running on the dask cluster successfully. However, the agent never reports any runs. Is there something else that is needed in order to connect the agent with flows executed on a dask cluster?

Hey, I have a complex data extraction flow with 2 main cases. Each case has its own tasks and some tasks are relevant to both of them. The layout of the flow looks somewhat like this -

with Flow('html_data_extraction__dev',
          storage=Docker(registry_url="us-central1-docker.pkg.dev/***/",
                         dockerfile="./Dockerfile"),
          schedule=daily_schedule, executor=LocalDaskExecutor(scheduler="processes")) as flow:
     mode = Parameter(name='mode', default=None)
     with case(mode, None):
          data_a=task_a
     with case(mode, 'onboard'):
          data_b=task_b
     
     data_c=merge(data_a, data_b)
     task_c
     task_d
      
     with case(mode, None):
          task_x
     with case(mode, 'onboard'):
          task_y

Tasks a and b retrieve some data (each one is relevant for a different data source), task c and d are common (not in a “case”) - doing X on the data (the data looks the same at this point) and then again tasks x and y are different - each one is relevant for a different case. When running locally (mac, flow,run…) it all works as expected. When running on Prefect cloud - all of the tasks gets skipped (exactly the same code and credentials) Any idea on what I’m missing here? I’m using “upstream_tasks” in order to run the tasks in a specific order when necessary. Thanks

hey guys I have a flow in prefect cloud that’s showing this error

{'_schema': 'Invalid data type: None'}

after it is scheduled but before the first task gets executed. When I try to run this flow in a local environment it executes as expected. Any idea what could be causing this issue?

Hello, I am looking into Prefect 2.0 and am wondering how we might implement a cleanup function - in prefect 1 from what I have seen we could use triggers "All Finished" when defining the task - I cant see the equivalent in V2, any suggestions?

I stumbled into the ECS Capacity Provider issue (PR just recently fixed https://github.com/PrefectHQ/prefect/pull/5411). I am curious when this might be released for Prefect 1.0 agents. This isn’t entirely a blocking issue for my team, but it is forcing us to workaround it via instance type selection via other methods.

Hi! One of my flows uses GQL to pull logs from another flow. The log pulling flow is directly triggered by the flow that produces the log to be pulled. I continue to get this error for the log pulling flow:

{'errors': [{'path': ['flow_run'], 'message': 'Operation timed out', 'extensions': {'code': 'API_ERROR'}}], 'data': None}

When re-running a flow at a later time, it runs successfully. Are there any docs that can provide info on API limits?

is there a write up/document about specifically what's new/different in prefect 2.0? dont really want to read through all of the docs from scratch, but a summary of what the changes are would be nice

Hello, my prefect 2.0 tests are going well...I see that Flow has a timeout_seconds parameter, is there any way to set Task level timeouts as well?

Does Prefect 1.1 support retries on

Flows

? Specifically, I’m trying to workaround some delayed scaling issues with ECS using EC2 instances (not ECS with Fargate tasks) Often, this failure is reported back to Prefect like the following error until Capacity Provider scaling has caught up again:

FAIL signal raised: FAIL('a4f09101-0577-41ce-b8b0-31b84f26d855 finished in state <Failed: "Failed to start task for flow run a4f09101-0577-41ce-b8b0-31b84f26d855. Failures: [{\'arn\': \'arn:aws:ecs:us-east-1:<redacted>:container-instance/a8bc98b7c6864874bc6d1138f758e8ea\', \'reason\': \'RESOURCE:CPU\'}]">')

I’m using the following calls to launch the sub-flows (as part of a larger script):

flow_a = create_flow_run(flow_name="A", project_name="myles")
wait_for_flow_a = wait_for_flow_run(flow_a, raise_final_state=True, stream_logs=True)

Hi there, I'm trying to implement some custom logic to handle retries. When a task times out or fails, I'd like to be able to spawn new tasks using the arguments passed to the original task. And if those tasks succeed, I'd like to be able to change the original task's state from failed to successful. Anyone have any advice? I've read the docs on state handlers and played around with it and haven't quite found what I'm looking for. Sample code here:

hey everyone, I'm trying to parellelize a flow that consists of several AirbyteConnectionTasks. Flow runs fine on the ECS Agent, but it runs each task in series. Is it possible to run the tasks in parallel when I'm generating them via a for loop, or is Prefect just running the tasks as they are generated? If so, how would this be done? Here is a toy code sample of what I'm working with

connections = ['conn1', 'conn2',
               'conn3']
               
with Flow("flow_name", run_config=RUN_CONFIG, storage=STORAGE, schedule=SCHEDULE) as flow:
    for conn_id in connections:
         flow.add_task( AirbyteConnectionTask(
            airbyte_server_host=<Airbyte host>
            airbyte_server_port=<airbyte port>,
            airbyte_api_version="v1",
            connection_id=conn_id
        )
         )
flow.run(executor=LocalDaskExecutor)

Hi There, I'm wondering exactly how the Prefect Docker Agent calls docker run. I have a

CMD

in my dockerfile and it seems to not be executing. That makes me think it is being overwritten. Maybe I have to provide an

ENTRYPOINT

intead?

Hi there, looking for some help running an ECS Agent with S3 storage. Getting a weird boto error.

Hi all, I am having a problem with creating an agent using prefect server as a backend. When I try to kubectl apply the agent manifest I get the following error:

prefect.exceptions.ClientError: Malformed response received from API.

Hello. Is there a resource on how to setup Prefect Server and Agents within AWS where Agents EC2 are in Private subnets and Server is setup on EC2 with Public subnet that covers security aspects in details?

Hi everybody, I'm working on a flow which should register another flow. And while flow.run() works perfectly fine for me, the running on the Cloud gives an error described in the picture. I've dived into this issue and realized that it does register another flow but outputs an error printing the success message))). So I tried changing print(msg) -> print(msg.encode('utf-8')) in the line 979 of the prefect/client/client.py and it helped me to fix this issue. Please consider adding this changing to the later versions of Prefect. Thank you in advance

Hi again, we have a requirement to handle two classes of errors: retryable errors and non-retryable errors. I think I saw some examples how this might be achieved somewhere but cant spot them for Prefect 2. In the non-retryable scenario we'd want the task state to go immediately to Failed (bypassing any retry settings configured on the task decorator)

I seemed like

prefect.tasks.kubernetes.job.RunNamespacedJob

doesn't report properly to prefect server. I didn't receive any log from spawned kubernetes job at all. Here are my screenshots

Hello everyone, I have a local agent which is not executing my flow run. The flow has been registered in the prefect server. The local hello world task can be executed locally but it would not update the state in the UI. If I run

prefect run -i "efc4568a-1fbd-4a14-8ef1-f137bdb45fcf" --watch

I receive the following warning:

WARNING | It has been 270 seconds and your flow run has not been submitted by an agent. Agent 122a102c-f083-48a8-b6f4-efd38e1b14f6 (agent) has matching labels and last queried a few seconds ago. It should deploy your flow run.

Hello, I’m new to Prefect and I need some help getting started. I’m currently struggling with two issues: 1. I’m setting up Prefect server on EKS using Helm. After

helm upgrade

ran successfully, I get commands such as

1 Run the following command to get the UI URL:

  UI_HOST=$( \
    kubectl get svc \
    --namespace prefect \
    --template "{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}" \
    prefect-server-ui \
  ) \
  && echo "UI available at: http://$UI_HOST:8080"...

when I run the above command, I get an error saying

error: error executing template "{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}": template: output:1:10: executing "output" at <index .status.loadBalancer.ingress 0>: error calling index: index of untyped nil

Then I have to run

kubectl port-forward <name of the service> <port>:<port>

to get ui and apollo servers running. Is there a way to run this without port forwarding? 2. I’m trying to run a hello world example flow on prefect server on my EKS cluster. I tried to trigger it from my command line by creating a project and running

prefect register --project test-project -p hello/

. The registering seem to run successfully but I don’t see the flow on the ui. I’m also not able to manually create flow project using ui. Am I interacting with two different prefect engine here? How do I register prefect flow so that it is running on my prefect server on EKS cluster?

Hi, I was wondering if anyone has tried using Redis as the storage backend for orion (prefect 2)? I see there is no built-in support to it. https://orion-docs.prefect.io/concepts/storage/ The closest we get is KV Server Storage. To use redis, does that mean I need to expose a HTTP API in between Orion and the redis server?