What would be the correct syntax to naming my tasks? The

name=""

suggested in the documentation isn't working for the format I have my tasks in

campaign_details_dbt_run = dbt_task(name="Campaign Details", command='dbt run -m campaign_details')

What's the difference between the signals.FAIL and signals.TRIGGERFAIL? Also, there's an option to raise the signals.RETRY but only retry the task after the specified retry_delay? If I call this inside a task it enters in an infinite loop.

Importing issues There was a post on this topic, but I can't find it... if I have a project with the following structure: -- project -- source_code -- test_code -- prefect_workflows it appears to be a problem to use in the prefect flow a functions from the

source_code

folder, that have local imports, and that we need to create a python package in order to use local code. is this correct or is there some way around it ?

dagster

Sorry. Please ignore.

Simple(ish) question, but how do you return the results of a

SnowflakeQuery

task into a Pandas dataframe?

If I encounter a certain condition in a flow run, I'd like to cancel the flow run (rather than fail it or mark it successful). It doesn't seem like there is a way to do this from prefect.engine.signals. Is there any way to do this (apart from the UI)? https://docs.prefect.io/core/concepts/execution.html#state-signals

Is there a way to get an alert if a connection to an agent becomes inactive ?

Hello, I have a flow that set DEFAULT retry greater than 0.

"PREFECT__TASKS__DEFAULTS__MAX_RETRIES": 2,
        "PREFECT__TASKS__DEFAULTS__RETRY_DELAY": 5,

Now, I want to set an individual task_A to

retry=0

(not to retry at all) and I find it impossible to do that. I will explain the details in the thread.

Hello, we use ECS to run our flows and lately I've been noticing that the task definition is created and immediately becomes

[INACTIVE]

. It's particularly strange because the inactive task definition shows "none" for the task role, but there is a role attached. I'll add a screenshot in the thread. The flow finishes successfully and writes to an s3 bucket that requires the role permissions. Any ideas what could be happening?

I am trying to use a data class as a paramter to my task but getting the below error while registering the flow. Error: " Beginning health checks... System Version check: OK Traceback (most recent call last): File "/opt/prefect/healthcheck.py", line 151, in <module> flows = cloudpickle_deserialization_check(flow_file_paths) File "/opt/prefect/healthcheck.py", line 44, in cloudpickle_deserialization_check flows.append(cloudpickle.loads(flow_bytes)) AttributeError: Can't get attribute 'WriteConfig' on <module 'pypline.prefect.tasks.hudi_writer_eks' from '/usr/local/lib/python3.8/site-packages/pypline/prefect/tasks/hudi_writer_eks.py'> " Prefect Code: I am trying to use a data class as a parameter to my task but getting the below error while registering the flow. I am using Prefect version 0.14.6 Error: " Beginning health checks... System Version check: OK Traceback (most recent call last): File "/opt/prefect/healthcheck.py", line 151, in <module> flows = cloudpickle_deserialization_check(flow_file_paths) File "/opt/prefect/healthcheck.py", line 44, in cloudpickle_deserialization_check flows.append(cloudpickle.loads(flow_bytes)) AttributeError: Can't get attribute 'WriteConfig' on <module 'pypline.prefect.tasks.hudi_writer_eks' from '/usr/local/lib/python3.8/site-packages/pypline/prefect/tasks/hudi_writer_eks.py'> import boto3 import prefect from kubernetes import client, config from kubernetes.client.rest import ApiException from prefect import task from prefect.engine.signals import FAIL, SKIP from pypline.lib.s3 import S3URI from dataclasses import dataclass @dataclass class WriteConfig: target_base_path: str target_table_name: str pk_column: str partition_column: str pre_combine_column: str source_path: str def validate(self): if not self.target_base_path: raise FAIL("'target_base_path' is required for hudi_options.") # TODO: validate that the required params are not empty @task def load_configs(executor_instances, hudi_options: WriteConfig): hudiparms = WriteConfig(target_base_path=hudi_options[0] ,target_table_name=hudi_options[1] , pk_column=hudi_options[2] , partition_column=hudi_options[3] , pre_combine_column=hudi_options[4] , source_path=hudi_options[5]) print(hudiparms) with Flow('Write Table',executor=get_dask_executor(for_k8, 4, 25, 'mmm/mmm-phase-0:' + mmm_version)) as hudi_flow:) as flow: executor_instances = Parameter('executor_instance', default=1) hudi_options= Parameter('hudi_options', default=[ "test_write" ,"test_table" ,"id" ,"one" ,"two" ,"userdata1.parquet"] ) load_configs(executor_instances, hudi_options)

I have a question regarding passing `LocalResult()`s in a flow of flows while testing the parent flow. The pattern here (subflow result passing) makes sense, but it seems like its designed for when running in prod on Server or Cloud. Is there a way to fudge a

flow_id

for

get_task_run_result

when just calling

flow.run()

for testing, or should I be thinking about this in an entirely different way when testing?

hello! beginner question: in DVC’s pipeline feature, a stage will only be re-executed if something in the stage definition or its dependencies changes (see https://dvc.org/doc/user-guide/project-structure/pipelines-files#stages). this is not default behaviour in prefect, right? how do i do this with prefect?

Hello everyone. We primarily use prefect for ETL jobs, loading data from one DB/file to our own DB. We use a master flow (flow of flows) which launches the sub-flows based on their dependency and scheduled_start_time. Sometimes, the systems we get our data from, report a delay by x hours. Is there any way (preferably through the UI) to pause this one flow and restart it later manually? Since there are more flows in the master flow, the other should continue as planned.

Is there something I am missing with the new

prefect run

CLI

--module

option? • Doing

python -m flows.myflow

works • Doing

prefect run -m flows.myflow

raises

No module named 'flows'

Hello! I have a workflow setup that works when I’m running on a local server. However, when I switch to production which is running on a KubernetesAgent, the flow fails at runtime with

Failed to load and execute Flow's environment: ValueError('Flow is not contained in this Storage')

I’m not sure how to debug this. What are the possible causes of this error?

Hello folks, while converting an old and monolithic ETL to a cool Prefect ETL, I stumbled upon the following situation:

@task("A small task"))
def small_task(some_params):
    # stuff happens here
    return something

@task("A complex task with many steps")
def complex_task(lots_of_params):
    # stuff happens
    # more stuff happens
    small_task_result = small_task(some_params)
    # even more stuff happens which uses small_task_result

My question is: what happens if I nest tasks as above? I suspect prefect would not like it, but I haven't set up a test infrastructure yet. Of course, I could use small_task as a simple function (remove the @task decorator), or break up complex_task in smaller tasks, but both solutions would mean for me to either leave things as they are (awful) or rewrite a lot of code.

Hello Prefect team, We had a connection timeout error with https://api.prefect.io earlier today around 2:20am Pacific time. Were there any known outages around then? Is there a page that tracks outages?

Does anyone by chance have a data engineering-specific cookiecutter recommendation? Most of the ones I’m seeing are for data science. Obviously I can make my own, but figured I’d see if other folks are using it

Hey Team, i'm having an issue where my jobs are failing to pick up "PREFECT__USER_CONFIG_PATH" environment variable that i've preset on a base docker image. But when i bash into the docker container while running, it works fine with flow.run() .. Can you help me understand what happens when a flow is deployed? We're currently using a modified gitlab storage that recursively downloads all files into the home directory, one of which is .prefect/config.toml. For example, whats the order of operations? My hypothesis is that the flow tries to run globals first, and then run the gitlab storage to get the flows, then follow tasks to execute? For more details, i'm trying to make the run_config assigned to the flow configurable through the config.toml file

# Registration Configs

flow.run_config = NomadRun(**prefect.config.run_config)

flow.storage = GitLab(path='flow.py', **prefect.config.storage)

Are we still thinking of doing this? We evaluated Pulumi a while back and it’s very cool; I’d be up for using this: https://github.com/PrefectHQ/server/pull/171

Hi community, has anyone used prefect to track data lineage? We have a project that would greatly benefit from knowing what data flows from where to where. Since prefect already builds a graph we have been thinking that we just need to put some effort in how we structure our flows so load and store happens in easily identified tasks, and then ask the right questions from the resulting DAG. If anyone had done something similar already it would be very interesting to hear how it went!

have a question on cloud standard offer what does it means by, 250 nightly ETL tasks, 10K production tasks, 100K ML tasks… aren’t all tasks just tasks? do we need to define a task as ETL, Production or ML ?

Suppose that in a flow of flows, flow A has a task that generates a "complex" object, e.g. a dataclass, that is then consumed by flow B as a parameter:

@task(result=LocalResult(...))
def a_do_something():
    return MyClass(1, 2, 3)

with Flow('A') as child_A:
    a_result = a_do_something() # 

with Flow('B') as child_B:
    b_param = Parameter('a_result_value')
    # option 3 below: b_param.serializer = MyClassJSONSerializer() or similar
    b_result = b_do_something(b_param)

with Flow('parent') as parent:
    a_id = create_flow_run(name=child_A.name)
    a_result = get_task_run_result(a_id, 'a_do_something')
    create_flow_run(name=child_B.name, parameters={'a_result_value': a_result})
    # also wait for a_id, set A upstream of B, wait for b_id

Because Parameters are/return? PrefectResults, they must be JSON serializable (indeed, if I just try the above I get a

Type Error: MyClass is not JSON serializable

or similar). Is the expectation that I should be messing with: 1. the serialization of LocalResult from

a_do_something

2. somehow injecting a serializer into Parameter (my attempts here have thus far failed - is this even possible?), or 3. modifying the

b_param

result (see the commented line in

child_B

flow) Note: based on my reading of several docs pages regardings Results and Serializers, option 2 strikes me as the most intuitive place to look. That is, it seems Prefect wants you to solve this there.

Hi gang, I am playing around with getting data logged to the stdout to be displayed in both the prefect logs and my team’s cloudwatch logs. Does anyone have any ideas how to do this? The flow is super simple but the logs aren’t showing up in cloudwatch:

from prefect.run_configs import ECSRun
from prefect import task, Flow
from prefect.storage import S3
import prefect

TASK_ARN = "an arn"
STORAGE = S3(bucket="my-cool-bucket")
LABELS = ["a label"]

RUN_CONFIG = ECSRun(
    labels=LABELS,
    task_role_arn=TASK_ARN,
    image="prefecthq/prefect:latest-python3.8",
    memory=512,
    cpu=256,
)


@task(log_stdout=True)
def bye_world():
    print("bye world")


@task(log_stdout=True)
def hello_world():
    print("Hello, World!")


with Flow("test_flow", run_config=RUN_CONFIG, storage=STORAGE) as flow:
    hello_world()
    bye_world()


flow.register(project_name="Tesla")

Hello. I'm struggling to get prefect to run/install at all. I find that the documentation is severely lacking in terms of having a useful guide on walking you through an install from scratch if you've never used Prefect before. I have a new Ubuntu 18.04 Server installation, docker and docker-compose installed, python 3.7 setup and set as the default. Run "pip install prefect" it's installed, the documentation says to now run "prefect server backend" or really "prefect" anything and I get nothing. Where are the steps in between to getting this running? https://github.com/prefecthq/prefect

Hi everyone, I'm having trouble with a custom

resource_manager

class, the flow times out and errors like this appear;

Error during execution of task: AttributeError("'NoneType' object has no attribute 'setup'")

Hello everyone! Hope you are all ok. I am trying to use the GraphQL API to get some information about our flow_runs. Basically what I need is some information about every flow_run and how long they took to finish (or has taken since creation until now for the case of the ones still running). The thing is that I'm very new to GraphQL and I'm not sure what I want is doable. It should be something like this:

query{
  flow_run{
    name,
    created,
    state,
    flow_id,
	duration: <here goes the flow_run execution duration>
  }
}

Ideally I could do something like

duration = end_time - start_time

, but I think GraphQL does not allow operations like these? I also checked the

flow_run_aggregate

query, but it does not contain such information. The real problem is that I want to consume this data from Grafana, so I have nowhere to run these calculations and the data should come already calculated in the response. Any ideas on how could I do this? Thanks a lot!

Hello everyone, there's a way to configure Prefect so he can use a proxy server? I'm trying to log in on Prefect cloud (using the CLI), but I'm receiving the following SSLError:

requests.exceptions.SSLError: HTTPSConnectionPool(host='<http://api.prefect.io|api.prefect.io>', port=443): Max retries exceeded with url: / (Caused by SSLError(SSLEOFError(8, 'EOF occurred in violation of protocol (_ssl.c:852)'),))