Hi, I’m trying to use the Snowflake Task to execute a “COPY INTO” query. It works fine with SELECT queries, but it just isn’t running the query when I switch it to a copy into and the task is coming back with TriggerFailed

Newbie question here - why can’t a single agent take care of various type of run deployments (Docker, Local, etc)?

I am trying to run two versions of a Flow on a Docker container using Docker Storage - the only (intended) difference is that one is built from tasks created via the @task decorator, and the other is built from tasks subclassed from the Task object. The decorator version runs fine, but the subclassed version results in this error when the Docker container is being built:

/opt/prefect/healthcheck.py:130: UserWarning: Flow uses module which is not importable. Refer to documentation on how to import custom modules <https://docs.prefect.io/api/latest/storage.html#docker>

flows = cloudpickle_deserialization_check(flow_file_paths)

Is there an obvious reason why this would happen? My tasks are defined in a module that I’m importing into the file where the Flows are defined. I can provide more info if needed!

Hi, what’s the correct way to use EnvVarSecret within a Flow, currently my code is below, but it tells me the task has not been added to the flow. If I don’t call it with

.run()

it complains that it is not JSON serializable:

with Flow(FLOW_NAME, storage=set_storage(FLOW_NAME), run_config=set_run_config(),) as flow:
    snowflake_password=EnvVarSecret(name='SNOWFLAKE_PASSWORD', raise_if_missing=True).run()
....
    upload_results = SnowflakeQuery(
        user=SNOWFLAKE_USER,
        password=snowflake_password,
        account=SNOWFLAKE_ACCOUNT,
        warehouse=SNOWFLAKE_WAREHOUSE,
        database=SNOWFLAKE_DATABASE,
        role=SNOWFLAKE_ROLE,
        schema=SNOWFLAKE_SCHEMA).map(query=queries)

I noticed that Prefect is a CNCF member but I didn't see it anywhere as a product in the Landscape. Did I miss something?

is there any mechanism in prefect for retrying a flow? I know i can set retries on tasks, but sometimes weird things happen with the k8s autoscaler and I am unable to assign a pod. In these cases, I want to backoff retry the entire flow, but I'm not finding where I would configure that. It never actually starts a task, so defining task retries doesn't help me here. I've tried defining retries specifically on the k8s job template in the past, but I've found that the prefect job will sometimes get confused on the state of the flow as its not directly managing the retries.

Hey all! @justabill, @Charleigh Liang and myself are going live in 10 minutes to chat about the history of dataflow automation. If you like reminiscing about old architecture decisions then this will be the stream for you! See you there on PrefectLive!

Hi again! If I’m using a prefect task (such as SnowflakeQuery or dbtshell) and want to set custom logging, how do I do that. For example, at the moment I have “Task ’SnowflakeQuery[52] Starting Run” …. “Task ’SnowflakeQuery[53] Starting Run” … “Task ’SnowflakeQuery[54] Starting Run” and instead I want to have “File load query file1.txt” ….. “File load query file2.text”. Is that possible? Can I just pass in names and mappings via the name task_args? Also, how do I set retries? Do I pass task_args too?

Newbie question - I am getting this error when I try to register my flow.

File "C:\etls\tractor_hours_analyzer\tractor_hours_analyzer.py", line 89, in <module>
    transformed = transform_add_files_to_dataframe(extracted)
    raise TypeError('too many positional arguments') from None
    TypeError: too many positional argumentswith

I am trying to use Github Actions for assigning storage to flows and registering them. Does it needs to know all the packages while doing that also?

Is it possible to set a Flow parameter’s type in code? When I go into one of our parametrized flows in Prefect Cloud, each parameter has a Type dropdown. I’d like to explicitly set each of our Parameters to one one the types from the dropdown.

Is it possible to create inline deployment instead of using CLI ? I am trying to run the Kubernetes tutorial, all looks good, but in my case I need to dynamically create flow, tasks, flows and deployment and want to create inline deployment. Is there any code example? PS : Question is for Orion.

Second question is about storage? When will the git storage available in Orion? I am trying to use s3 storage but it uses bucket root directory for deployments, I can not provide a sub folder in bucket.

Third question is about kubernetes orion installation. Everytime I started my docker/kubernetes environment, services become active but I have to reconfigure storage, create work queues and all delpyment/flows are gone in UI. Isn’t the data or the POD disk persisted ?

Still coming to Databricks and I guess in general with Tasks. Databricks has added git support for jobs, and the current DatabricksSubmitMultitaskRun doesn't support it. I am in doubt among the possible approaches: • Create a DatabricksSubmitMultitaskRun custom implementation. However, I will access underscore methods in prefect package • Use the databricks CLI python library to create a job and then just run it via Prefect • Other? The real problem is that the Task does not allow dependency injection (i.e. the databricks client is created within the Run function, so it's not easy to override it). I guess the design of the Task is concerning in the sense that is not extensible, one needs to rewrite it from scratch

In old-school Prefect (not Orion), is there a workaround for adding build args (i.e. Dockerfile's ARG instead of ENV)? Currently Prefect's

Docker

class takes an

env_vars

dict, but no

build_args

dict, which would be nice. Here is the place in the

Docker

class where ENVs are generated: https://github.com/PrefectHQ/prefect/blob/8e04ccad071d1127afc7ca3580f1fe6f4e884f27/src/prefect/storage/docker.py#L437-L442 ...the ARGs could go right above there. In any case, ultimately what I'm trying to do is get the

Docker

class to install

python_dependencies

from our private PyPi server. Is there a good way to do that?

Thread about loading flows from modules instead of .py files or cloudpickle-serialized blobs =>

I have registered the flows via GitHub Actions and the agent is there. When I run the flow, it looks like it got stuck. Can someone help me on debugging this? or what is happening here?

Morning All We are using prefect via a kubernetes agent. We've noticed that failed flows don't clean up prefect jobs and in some cases the dask-excecutor. Is there any passive mechanism to clean up these dead jobs ?

etl                prefect-job-29a6d24a-hwnzx                         0/1     ImagePullBackOff             0          6d20h
etl                prefect-job-30bd9ad7-zsc6z                         0/1     CreateContainerConfigError   0          11h
etl                prefect-job-39468921-lt95t                         0/1     CreateContainerConfigError   0          6d20h
etl                prefect-job-3c1aee81-hvbv4                         0/1     Completed                    0          7h49m
etl                prefect-job-3c84d152-wfqw9                         0/1     Completed                    0          7h53m
etl                prefect-job-3fb2397e-7dc5b                         0/1     ImagePullBackOff             0          11h
etl                prefect-job-5ba165eb-rfxhq                         0/1     CreateContainerConfigError   0          6d20h
etl                prefect-job-62ae93a3-ptbcr                         0/1     CreateContainerConfigError   0          6d20h
etl                prefect-job-6cfd964c-p7xpl                         0/1     ImagePullBackOff             0          6d20h
etl                prefect-job-86e07407-nmpwb                         0/1     ImagePullBackOff             0          6d20h
etl                prefect-job-9478c771-bfg26                         0/1     ImagePullBackOff             0          11h
etl                prefect-job-950ba45b-tv4f2                         0/1     CreateContainerConfigError   0          11h
etl                prefect-job-a199b1d6-6vwmd                         0/1     Completed                    0          7h49m
etl                prefect-job-a4976ce3-gxqv6                         0/1     Completed                    0          6h5m

Good morning prefect community 😊. We are evaluating a switch to either Prefect or Dagster. On the surface the feature sets look quite similar. Do you have any resources aiding with the decision which solution to choose?

Morning, we've set up new Prefect env in K8s, but some old dataflow have DockerRun() as run_config. These dataflows are failing with error: "Flow run 46a30aa1-11f0-4c77-bb56-afab879db8bc has a

run_config

of type

DockerRun

, only

KubernetesRun

is supported" Suppose that Prefect knows it runs in K8s and now all dataflow should use K8s agent, right?

Hi All We are using prefect server 0.15.7 and running the flows on yarn Cluster. I have 20 mapped tasks and each mapped task loops over a list , does some processing and saves output to hadoop. I am getting the below error intermittently. Any help would be appreciated.

While registering the flow it is not storing the configs file which is present in a different directory. But some modules needs them. How to solve this?

main-project
  | 
  src
     |
     | __init__.py
     | module1.py
     | module2.py
  |
  configs
     | 
     | config.yaml
  |
  flows
     |
     | sample_flow.py

Hi all, we are using

prefect server

for local development and

prefect cloud

for production which is quite neat in terms of speed. When doing this we set the --external-postgres flag in order to use our development database, which works great! However we are using Hasura for our other backend service which creates some git-pain as Prefect overwrites our local Hasura metadata files. Have anyone run into the same problem ?

Hi again, theoretical question. If most use Docker/K8s to run dataflow on (each dataflow has own image, or each flow executes common image), why do we need Agent at all?

Hey, i’m running a prefect agent on EKS / Fargate. It’s been working great for a while now, spinning up a new pod, running and deleting it. I think for some reason something got changed on the AWS account and the spinning up of pods failed. However, it kept trying to retry and spin up a new pod, and the old pod didn’t get cleaned up. Now whenever I manually try to delete the pod (kubectl delete pods) it just recreates instantly. So now i have like 6 pods that are stuck in Pending status that I can’t delete. Any thoughts on how to clean everything up?

How would you describe Prefect in a compact way that highlights its features? It's not a workflow manager in the sense that it is not Camunda, it is not Temporal, it is not AWS Step. It's not Conductor by Netflix

While

prefect_test_harness

provides an environment for flows to be tested, how might I go about testing a task in Prefect 2.0? It seems like calling

.run()

on tasks has not made it over from 1.0 yet.

I am struggling a little with Prefect programming patterns. Take this example from this doc: https://docs.prefect.io/core/advanced_tutorials/task-looping.html

with Flow("mapped-fibonacci") as mapped_flow:
    ms = Parameter("ms")
    fib_nums = compute_large_fibonacci.map(ms)

flow_state = mapped_flow.run(ms=[10, 100, 1000, 1500])

nice but what about non array parameters? Let's say I have two parameters, one that's an array, and the other which is a constant.

with Flow("mapped-fibonacci") as mapped_flow:
    ms = Parameter("ms")
    ms2 = Parameter("ms2")
    fib_nums = compute_large_fibonacci.map(ms, ms2)

flow_state = mapped_flow.run(ms=[10, 100, 1000, 1500])

Hi guys ! I’m trying to setup a

Prefect Agent

and a

prefect server

on my local computer, everything residing inside of

containers

. My

Prefect Server

`container(s)`is based on the

docker-compose

file retrieved from

prefect server config > server-docker-compose.yaml

. When spinning up a container for an Agent and running

prefect agent local start --api <http://host.docker.internal:4200>

, it says that I must be authenticated to communicate with the containerized server living in my local machine. My first question is: Is

authentication

mandatory in this kind of situation , for local run ? When I authenticate by creating a new tenant using

prefect server create-tenant -n default

outside the Agent container, I’m able to spin up the

Agent Container

and run everything. The problem is that I wanted to run the

create-tenant

inside the container too, but it’s not possible as I’m getting connection refused between containers.